iroh_db_core/
schema.rs

1use std::collections::BTreeSet;
2
3use minicbor::{Decoder, Encoder};
4
5use crate::{CollectionId, SchemaId};
6
7const MAGIC: &str = "iroh-db";
8const FORMAT_MAJOR: u16 = 1;
9const FORMAT_MINOR: u16 = 0;
10const SCHEMA_KIND: u8 = 1;
11const MAX_NAME_LEN: usize = 255;
12const MAX_TYPE_NAME_LEN: usize = 1_024;
13const MAX_FIELDS: usize = 4_096;
14
15/// A built-in deterministic replicated-field strategy.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
17#[repr(u8)]
18pub enum CrdtKind {
19    /// Causal register with a deterministic concurrent-write tie break.
20    Lww = 1,
21    /// Add-wins observed-remove set.
22    OrSet = 2,
23    /// Set that can only grow.
24    GrowOnlySet = 3,
25    /// Per-author positive/negative counter.
26    PnCounter = 4,
27    /// Register retaining every causally maximal value.
28    MultiValue = 5,
29    /// Write-once value that preserves concurrent conflicts.
30    Immutable = 6,
31    /// Replicated growable ordered list.
32    OrderedList = 7,
33}
34
35impl TryFrom<u8> for CrdtKind {
36    type Error = SchemaError;
37
38    fn try_from(value: u8) -> Result<Self, Self::Error> {
39        match value {
40            1 => Ok(Self::Lww),
41            2 => Ok(Self::OrSet),
42            3 => Ok(Self::GrowOnlySet),
43            4 => Ok(Self::PnCounter),
44            5 => Ok(Self::MultiValue),
45            6 => Ok(Self::Immutable),
46            7 => Ok(Self::OrderedList),
47            value => Err(SchemaError::UnknownCrdtKind(value)),
48        }
49    }
50}
51
52/// The stable replicated contract for one record field.
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct FieldDescriptor {
55    id: u32,
56    name: String,
57    crdt: CrdtKind,
58    value_type: String,
59    indexed: bool,
60    record_id: bool,
61}
62
63impl FieldDescriptor {
64    /// Declares a replicated field.
65    pub fn new(
66        id: u32,
67        name: impl Into<String>,
68        crdt: CrdtKind,
69        value_type: impl Into<String>,
70    ) -> Self {
71        Self {
72            id,
73            name: name.into(),
74            crdt,
75            value_type: value_type.into(),
76            indexed: false,
77            record_id: false,
78        }
79    }
80
81    /// Marks this field as a materialized secondary index.
82    #[must_use]
83    pub const fn indexed(mut self) -> Self {
84        self.indexed = true;
85        self
86    }
87
88    /// Marks this field as the unique record identifier.
89    #[must_use]
90    pub const fn record_id(mut self) -> Self {
91        self.record_id = true;
92        self
93    }
94
95    /// Returns the stable numeric field identifier.
96    pub const fn id(&self) -> u32 {
97        self.id
98    }
99
100    /// Returns the declared Rust field name.
101    pub fn name(&self) -> &str {
102        &self.name
103    }
104
105    /// Returns the replicated merge strategy.
106    pub const fn crdt(&self) -> CrdtKind {
107        self.crdt
108    }
109
110    /// Returns the stable logical value-type name.
111    pub fn value_type(&self) -> &str {
112        &self.value_type
113    }
114
115    /// Returns whether the field has a secondary index.
116    pub const fn is_indexed(&self) -> bool {
117        self.indexed
118    }
119
120    /// Returns whether the field is the record identifier.
121    pub const fn is_record_id(&self) -> bool {
122        self.record_id
123    }
124}
125
126/// A validated, versioned collection schema.
127#[derive(Debug, Clone, PartialEq, Eq)]
128pub struct SchemaDescriptor {
129    collection_name: String,
130    collection_id: CollectionId,
131    version: u32,
132    fields: Vec<FieldDescriptor>,
133}
134
135impl SchemaDescriptor {
136    /// Validates and constructs a collection schema.
137    pub fn new(
138        collection_name: impl Into<String>,
139        version: u32,
140        mut fields: Vec<FieldDescriptor>,
141    ) -> Result<Self, SchemaError> {
142        let collection_name = collection_name.into();
143        validate_name("collection", &collection_name, MAX_NAME_LEN)?;
144        if fields.len() > MAX_FIELDS {
145            return Err(SchemaError::TooManyFields(fields.len()));
146        }
147
148        fields.sort_by_key(FieldDescriptor::id);
149        let mut ids = BTreeSet::new();
150        let mut names = BTreeSet::new();
151        let mut record_ids = 0_usize;
152
153        for field in &fields {
154            validate_name("field", field.name(), MAX_NAME_LEN)?;
155            validate_name("value type", field.value_type(), MAX_TYPE_NAME_LEN)?;
156            if !ids.insert(field.id()) {
157                return Err(SchemaError::DuplicateFieldId(field.id()));
158            }
159            if !names.insert(field.name().to_owned()) {
160                return Err(SchemaError::DuplicateFieldName(field.name().to_owned()));
161            }
162            record_ids += usize::from(field.is_record_id());
163        }
164
165        match record_ids {
166            0 => return Err(SchemaError::MissingRecordId),
167            1 => {}
168            _ => return Err(SchemaError::MultipleRecordIds),
169        }
170
171        let collection_id = CollectionId::for_name(&collection_name);
172        Ok(Self {
173            collection_name,
174            collection_id,
175            version,
176            fields,
177        })
178    }
179
180    /// Returns the human-readable declared collection name.
181    pub fn collection_name(&self) -> &str {
182        &self.collection_name
183    }
184
185    /// Returns the stable collection identifier.
186    pub const fn collection_id(&self) -> CollectionId {
187        self.collection_id
188    }
189
190    /// Returns the application-controlled schema version.
191    pub const fn version(&self) -> u32 {
192        self.version
193    }
194
195    /// Returns fields in canonical numeric-ID order.
196    pub fn fields(&self) -> &[FieldDescriptor] {
197        &self.fields
198    }
199
200    /// Returns the domain-separated hash of the canonical descriptor.
201    pub fn schema_id(&self) -> SchemaId {
202        let mut hasher = blake3::Hasher::new();
203        hasher.update(b"iroh-db/schema/v1");
204        hasher.update(&self.encode_canonical());
205        SchemaId::from_bytes(*hasher.finalize().as_bytes())
206    }
207
208    /// Encodes the descriptor into its durable canonical representation.
209    pub fn encode_canonical(&self) -> Vec<u8> {
210        let mut encoder = Encoder::new(Vec::new());
211        encode_schema(&mut encoder, self).expect("encoding into Vec cannot fail");
212        encoder.into_writer()
213    }
214
215    /// Decodes and validates an exact canonical descriptor.
216    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, SchemaError> {
217        let decoded = decode_schema(bytes)?;
218        if decoded.encode_canonical() != bytes {
219            return Err(SchemaError::NonCanonicalEncoding);
220        }
221        Ok(decoded)
222    }
223}
224
225/// Schema validation or decoding failure.
226#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
227pub enum SchemaError {
228    /// A required name was empty or exceeded its byte limit.
229    #[error("invalid {kind} name length {length}; maximum is {maximum}")]
230    InvalidName {
231        /// Name category.
232        kind: &'static str,
233        /// Actual UTF-8 byte length.
234        length: usize,
235        /// Maximum accepted UTF-8 byte length.
236        maximum: usize,
237    },
238    /// More fields were supplied than the decoder permits.
239    #[error("schema contains too many fields: {0}")]
240    TooManyFields(usize),
241    /// Two fields use the same stable numeric ID.
242    #[error("duplicate field id {0}")]
243    DuplicateFieldId(u32),
244    /// Two fields use the same declared name.
245    #[error("duplicate field name {0}")]
246    DuplicateFieldName(String),
247    /// No record identifier was declared.
248    #[error("schema must declare exactly one record id")]
249    MissingRecordId,
250    /// More than one record identifier was declared.
251    #[error("schema declares more than one record id")]
252    MultipleRecordIds,
253    /// The encoded merge strategy is unknown.
254    #[error("unknown CRDT kind {0}")]
255    UnknownCrdtKind(u8),
256    /// The object uses an unsupported format version or kind.
257    #[error("unsupported schema format")]
258    UnsupportedFormat,
259    /// A collection ID did not match its declared name.
260    #[error("collection id does not match the declared name")]
261    CollectionIdMismatch,
262    /// The CBOR object was malformed or violated a decode limit.
263    #[error("invalid schema encoding: {0}")]
264    InvalidEncoding(String),
265    /// The object was valid CBOR but not the unique canonical byte representation.
266    #[error("schema encoding is not canonical")]
267    NonCanonicalEncoding,
268}
269
270fn validate_name(kind: &'static str, value: &str, maximum: usize) -> Result<(), SchemaError> {
271    let length = value.len();
272    if length == 0 || length > maximum {
273        return Err(SchemaError::InvalidName {
274            kind,
275            length,
276            maximum,
277        });
278    }
279    Ok(())
280}
281
282fn encode_schema(
283    encoder: &mut Encoder<Vec<u8>>,
284    schema: &SchemaDescriptor,
285) -> Result<(), minicbor::encode::Error<std::convert::Infallible>> {
286    encoder.map(8)?;
287    encoder.u8(0)?.str(MAGIC)?;
288    encoder.u8(1)?.u16(FORMAT_MAJOR)?;
289    encoder.u8(2)?.u16(FORMAT_MINOR)?;
290    encoder.u8(3)?.u8(SCHEMA_KIND)?;
291    encoder.u8(4)?.str(schema.collection_name())?;
292    encoder.u8(5)?.bytes(schema.collection_id().as_bytes())?;
293    encoder.u8(6)?.u32(schema.version())?;
294    encoder.u8(7)?.array(schema.fields().len() as u64)?;
295    for field in schema.fields() {
296        encoder.map(6)?;
297        encoder.u8(0)?.u32(field.id())?;
298        encoder.u8(1)?.str(field.name())?;
299        encoder.u8(2)?.u8(field.crdt() as u8)?;
300        encoder.u8(3)?.str(field.value_type())?;
301        encoder.u8(4)?.bool(field.is_indexed())?;
302        encoder.u8(5)?.bool(field.is_record_id())?;
303    }
304    Ok(())
305}
306
307fn decode_schema(bytes: &[u8]) -> Result<SchemaDescriptor, SchemaError> {
308    let mut decoder = Decoder::new(bytes);
309    expect_map_len(&mut decoder, 8)?;
310    expect_key(&mut decoder, 0)?;
311    if decoder.str().map_err(decode_error)? != MAGIC {
312        return Err(SchemaError::UnsupportedFormat);
313    }
314    expect_key(&mut decoder, 1)?;
315    if decoder.u16().map_err(decode_error)? != FORMAT_MAJOR {
316        return Err(SchemaError::UnsupportedFormat);
317    }
318    expect_key(&mut decoder, 2)?;
319    if decoder.u16().map_err(decode_error)? != FORMAT_MINOR {
320        return Err(SchemaError::UnsupportedFormat);
321    }
322    expect_key(&mut decoder, 3)?;
323    if decoder.u8().map_err(decode_error)? != SCHEMA_KIND {
324        return Err(SchemaError::UnsupportedFormat);
325    }
326    expect_key(&mut decoder, 4)?;
327    let collection_name = decoder.str().map_err(decode_error)?.to_owned();
328    expect_key(&mut decoder, 5)?;
329    let encoded_collection_id = decoder.bytes().map_err(decode_error)?;
330    let encoded_collection_id: [u8; 32] = encoded_collection_id
331        .try_into()
332        .map_err(|_| SchemaError::InvalidEncoding("collection id must be 32 bytes".into()))?;
333    expect_key(&mut decoder, 6)?;
334    let version = decoder.u32().map_err(decode_error)?;
335    expect_key(&mut decoder, 7)?;
336    let field_count = decoder
337        .array()
338        .map_err(decode_error)?
339        .ok_or_else(|| SchemaError::InvalidEncoding("indefinite field array".into()))?;
340    let field_count = usize::try_from(field_count)
341        .map_err(|_| SchemaError::InvalidEncoding("field count overflow".into()))?;
342    if field_count > MAX_FIELDS {
343        return Err(SchemaError::TooManyFields(field_count));
344    }
345
346    let mut fields = Vec::with_capacity(field_count);
347    for _ in 0..field_count {
348        expect_map_len(&mut decoder, 6)?;
349        expect_key(&mut decoder, 0)?;
350        let id = decoder.u32().map_err(decode_error)?;
351        expect_key(&mut decoder, 1)?;
352        let name = decoder.str().map_err(decode_error)?.to_owned();
353        expect_key(&mut decoder, 2)?;
354        let crdt = CrdtKind::try_from(decoder.u8().map_err(decode_error)?)?;
355        expect_key(&mut decoder, 3)?;
356        let value_type = decoder.str().map_err(decode_error)?.to_owned();
357        expect_key(&mut decoder, 4)?;
358        let indexed = decoder.bool().map_err(decode_error)?;
359        expect_key(&mut decoder, 5)?;
360        let record_id = decoder.bool().map_err(decode_error)?;
361
362        let mut field = FieldDescriptor::new(id, name, crdt, value_type);
363        if indexed {
364            field = field.indexed();
365        }
366        if record_id {
367            field = field.record_id();
368        }
369        fields.push(field);
370    }
371
372    if decoder.position() != bytes.len() {
373        return Err(SchemaError::InvalidEncoding("trailing data".into()));
374    }
375
376    let schema = SchemaDescriptor::new(collection_name, version, fields)?;
377    if schema.collection_id().as_bytes() != &encoded_collection_id {
378        return Err(SchemaError::CollectionIdMismatch);
379    }
380    Ok(schema)
381}
382
383fn expect_map_len(decoder: &mut Decoder<'_>, expected: u64) -> Result<(), SchemaError> {
384    let actual = decoder
385        .map()
386        .map_err(decode_error)?
387        .ok_or_else(|| SchemaError::InvalidEncoding("indefinite map".into()))?;
388    if actual != expected {
389        return Err(SchemaError::InvalidEncoding(format!(
390            "expected map length {expected}, got {actual}"
391        )));
392    }
393    Ok(())
394}
395
396fn expect_key(decoder: &mut Decoder<'_>, expected: u8) -> Result<(), SchemaError> {
397    let actual = decoder.u8().map_err(decode_error)?;
398    if actual != expected {
399        return Err(SchemaError::InvalidEncoding(format!(
400            "expected key {expected}, got {actual}"
401        )));
402    }
403    Ok(())
404}
405
406// `Result::map_err` requires ownership even though formatting only borrows the error.
407#[allow(clippy::needless_pass_by_value)]
408fn decode_error(error: minicbor::decode::Error) -> SchemaError {
409    SchemaError::InvalidEncoding(error.to_string())
410}
iroh_db_core/schema.rs

iroh_db_core/
schema.rs
