harmos_domain/inventory/
publications.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::collections::BTreeMap;
18// Third Party Imports
19use serde::Deserialize;
20// Local Imports
21use super::{StreamDeclaration, digits, quoted, quoted_len, write};
22use crate::config::SchemaSettings;
23use crate::schema::declaration::{Definition, Named};
24use crate::schema::graph::Schema;
25
26/* === Constants === */
27
28const HEAD: &[u8] = b"\0HARMOS:ARTIFACT:V1:{\"kind\":\"publications\",\"facility\":";
29const OWNER: &[u8] = b",\"owner\":";
30const STREAMS: &[u8] = b",\"streams\":[";
31const ID: &[u8] = b"{\"id\":";
32const SCHEMA: &[u8] = b",\"schema\":";
33const WINDOW: &[u8] = b",\"window\":";
34const METADATA: &[u8] = b",\"metadata\":";
35const ROW_METADATA: &[u8] = b",\"row_metadata\":";
36const ROW: &[u8] = b",\"row\":";
37const TAIL: &[u8] = b"]}\0";
38const METADATA_ID: &[u8] = b"{\"identifier\":";
39const METADATA_LABEL: &[u8] = b",\"label\":";
40
41/* === Macros === */
42
43/* === Type Aliases === */
44
45/* === Structs === */
46
47#[derive(Clone, Copy)]
48pub struct Entry {
49    pub id: &'static str,
50    pub metadata_json: &'static str,
51    pub row_metadata_json: &'static str,
52    pub schema: u32,
53    pub window: usize,
54    pub row: Definition,
55}
56
57#[derive(Deserialize)]
58#[serde(deny_unknown_fields)]
59pub struct Manifest {
60    pub facility: String,
61    pub owner: String,
62    pub streams: Vec<StreamDeclaration>,
63}
64
65#[derive(Deserialize)]
66#[serde(deny_unknown_fields)]
67pub struct NamedStream {
68    pub metadata: crate::Metadata,
69
70    #[serde(default)]
71    pub row_metadata: Vec<crate::StreamField>,
72
73    pub id: String,
74    pub schema: u32,
75    pub window: usize,
76    pub row: String,
77}
78
79#[derive(Deserialize)]
80#[serde(deny_unknown_fields)]
81pub struct NamedManifest {
82    pub facility: String,
83    pub owner: String,
84    pub streams: Vec<NamedStream>,
85}
86
87/* === Enums === */
88
89#[derive(Debug, thiserror::Error)]
90pub enum Error {
91    #[error("publication inventory differs from authored streams")]
92    Mismatch,
93
94    #[error("the artifact carries no complete {facility} publication inventory for `{owner}`")]
95    Missing { facility: String, owner: String },
96
97    #[error("the artifact carries multiple complete {facility} publication inventories for `{owner}`")]
98    Duplicate { facility: String, owner: String },
99
100    #[error("`{id}` is not a portable Harmos stream id")]
101    InvalidId { id: String },
102
103    #[error("the stream `{id}` has a zero schema ordinal")]
104    ZeroSchema { id: String },
105
106    #[error("the stream `{id}` has a zero window")]
107    ZeroWindow { id: String },
108
109    #[error(
110        "the stream `{id}` is declared more than once in its publication inventory; each stream must appear exactly once in the artifact publication grant"
111    )]
112    DuplicateStream { id: String },
113
114    #[error("stream `{id}` has invalid presentation metadata: {reason}")]
115    Metadata { id: String, reason: String },
116
117    #[error("stream `{id}` has an invalid field schema: {source}")]
118    Fields {
119        id: String,
120
121        #[source]
122        source: crate::schema::Error,
123    },
124}
125
126/* === Traits === */
127
128pub trait StreamSpec {
129    type Row;
130    const ID: &'static str;
131    const VERSION: u32;
132    const WINDOW: usize;
133    const METADATA_JSON: &'static str = "";
134    const ROW_METADATA_JSON: &'static str = "[]";
135
136    fn metadata() -> crate::Metadata {
137        if Self::METADATA_JSON.is_empty() {
138            crate::Metadata::new(crate::Identifier::declared(Self::ID, "stream"), Self::ID, None)
139        } else {
140            serde_json::from_str(Self::METADATA_JSON).expect("macro-generated stream metadata is valid")
141        }
142    }
143
144    fn row_metadata() -> Vec<crate::StreamField> {
145        serde_json::from_str(Self::ROW_METADATA_JSON).expect("macro-generated row metadata is valid")
146    }
147}
148
149/* === Implementations === */
150
151impl NamedManifest {
152    pub fn resolve(self, declarations: &[Named], settings: &mut SchemaSettings) -> Result<Manifest, Error> {
153        let streams = self
154            .streams
155            .into_iter()
156            .map(|stream| {
157                let row = Schema::from_named(&stream.row, declarations, settings).map_err(|source| Error::Fields {
158                    id: stream.id.clone(),
159                    source: source.into(),
160                })?;
161                let row = super::retain_schema(row, settings).map_err(|source| Error::Fields {
162                    id: stream.id.clone(),
163                    source: source.into(),
164                })?;
165                let stream = StreamDeclaration {
166                    metadata: stream.metadata,
167                    row_metadata: stream.row_metadata,
168                    id: stream.id,
169                    schema: stream.schema,
170                    window: stream.window,
171                    row,
172                };
173                validate(&stream)?;
174                Ok(stream)
175            })
176            .collect::<Result<_, Error>>()?;
177        Ok(Manifest {
178            facility: self.facility,
179            owner: self.owner,
180            streams,
181        })
182    }
183}
184
185impl Entry {
186    pub fn metadata(&self) -> crate::Metadata {
187        if self.metadata_json.is_empty() {
188            crate::Metadata::new(crate::Identifier::declared(self.id, "stream"), self.id, None)
189        } else {
190            serde_json::from_str(self.metadata_json).expect("declared stream metadata is valid")
191        }
192    }
193
194    pub fn row_metadata(&self) -> Vec<crate::StreamField> {
195        serde_json::from_str(self.row_metadata_json).expect("declared row metadata is valid")
196    }
197
198    pub const fn of<S: StreamSpec>() -> Self
199    where
200        S::Row: crate::schema::Described,
201    {
202        Self {
203            id: S::ID,
204            metadata_json: S::METADATA_JSON,
205            row_metadata_json: S::ROW_METADATA_JSON,
206            schema: S::VERSION,
207            window: S::WINDOW,
208            row: <S::Row as crate::schema::Described>::DEFINITION,
209        }
210    }
211}
212
213/* === Public Functions === */
214
215pub const fn merge<const N: usize>(named: &[Entry], sampled: &[Entry]) -> [Entry; N] {
216    let mut merged = [Entry {
217        id: "",
218        metadata_json: "",
219        row_metadata_json: "[]",
220        schema: 0,
221        window: 0,
222        row: Definition::of::<()>(),
223    }; N];
224    let mut index = 0;
225    while index < named.len() {
226        merged[index] = named[index];
227        index += 1;
228    }
229    let mut offset = 0;
230    while offset < sampled.len() {
231        merged[index + offset] = sampled[offset];
232        offset += 1;
233    }
234
235    merged
236}
237
238pub const fn require_tree_subset(observers: &[crate::catalog::Node<super::observers::Entry>], grants: &[Entry]) {
239    let mut index = 0;
240    while index < observers.len() {
241        match &observers[index] {
242            crate::catalog::Node::Leaf(entry) => require_subset(&[entry.streams], grants),
243            crate::catalog::Node::Nested(nested) => require_tree_subset(nested, grants),
244        }
245        index += 1;
246    }
247}
248
249pub const fn require_subset(outputs: &[&[Entry]], grants: &[Entry]) {
250    let mut observer = 0;
251    while observer < outputs.len() {
252        let mut output = 0;
253        while output < outputs[observer].len() {
254            let required = &outputs[observer][output];
255            let mut grant = 0;
256            let mut found = false;
257            while grant < grants.len() {
258                let declared = &grants[grant];
259                if same_bytes(required.id, declared.id) && required.schema == declared.schema && required.window == declared.window {
260                    found = true;
261                    break;
262                }
263                grant += 1;
264            }
265            assert!(found, "observer outputs must be a subset of the artifact publication grant");
266            output += 1;
267        }
268        observer += 1;
269    }
270}
271
272pub fn select(manifests: Vec<Manifest>, facility: &str, owner: &str) -> Result<Vec<StreamDeclaration>, Error> {
273    let mut matching = manifests.into_iter().filter(|m| m.facility == facility && m.owner == owner);
274    let manifest = matching.next().ok_or_else(|| Error::Missing {
275        facility: facility.into(),
276        owner: owner.into(),
277    })?;
278    if matching.next().is_some() {
279        return Err(Error::Duplicate {
280            facility: facility.into(),
281            owner: owner.into(),
282        });
283    }
284    let mut streams = BTreeMap::new();
285    for stream in manifest.streams {
286        validate(&stream)?;
287        let id = stream.id.clone();
288        if streams.insert(id.clone(), stream).is_some() {
289            return Err(Error::DuplicateStream { id });
290        }
291    }
292    Ok(streams.into_values().collect())
293}
294
295pub fn validate_match(expected: &[Entry], actual: &[StreamDeclaration]) -> Result<(), Error> {
296    let mut authored = BTreeMap::new();
297    for entry in expected {
298        let metadata = if entry.metadata_json.is_empty() {
299            let id = crate::Identifier::new(entry.id).map_err(|_| Error::InvalidId { id: entry.id.to_owned() })?;
300            crate::Metadata::new(id, entry.id, None)
301        } else {
302            serde_json::from_str(entry.metadata_json).map_err(|source| Error::Metadata {
303                id: entry.id.to_owned(),
304                reason: source.to_string(),
305            })?
306        };
307        let row_metadata = serde_json::from_str(entry.row_metadata_json).map_err(|source| Error::Metadata {
308            id: entry.id.to_owned(),
309            reason: source.to_string(),
310        })?;
311        let stream = StreamDeclaration {
312            metadata,
313            row_metadata,
314            id: entry.id.to_owned(),
315            schema: entry.schema,
316            window: entry.window,
317            row: Schema::from_declaration(entry.row, &SchemaSettings::default()).map_err(|source| Error::Fields {
318                id: entry.id.to_owned(),
319                source: source.into(),
320            })?,
321        };
322        validate(&stream)?;
323        if authored.insert(entry.id, stream).is_some() {
324            return Err(Error::DuplicateStream { id: entry.id.to_owned() });
325        }
326    }
327    let mut admitted = BTreeMap::new();
328    for stream in actual {
329        validate(stream)?;
330        if admitted.insert(stream.id.as_str(), stream).is_some() {
331            return Err(Error::DuplicateStream { id: stream.id.clone() });
332        }
333    }
334    if authored.len() != admitted.len() || authored.iter().any(|(id, stream)| admitted.get(id).is_none_or(|actual| *actual != stream)) {
335        return Err(Error::Mismatch);
336    }
337    Ok(())
338}
339
340pub fn validate(stream: &StreamDeclaration) -> Result<(), Error> {
341    if !crate::stream_subset(&stream.id) {
342        return Err(Error::InvalidId { id: stream.id.clone() });
343    }
344    if stream.schema == 0 {
345        return Err(Error::ZeroSchema { id: stream.id.clone() });
346    }
347    if stream.window == 0 {
348        return Err(Error::ZeroWindow { id: stream.id.clone() });
349    }
350    if stream.metadata.identifier.as_str() != stream.id {
351        return Err(Error::Metadata {
352            id: stream.id.clone(),
353            reason: "identifier differs from stream identity".into(),
354        });
355    }
356    stream.row.prepare(&SchemaSettings::default()).map_err(|source| Error::Fields {
357        id: stream.id.clone(),
358        source: source.into(),
359    })?;
360    let mut described = std::collections::BTreeSet::new();
361    for field in &stream.row_metadata {
362        let name = field.metadata.identifier.as_str();
363        if !stream.row.definitions[stream.row.root].iter().any(|wire| wire.name == name) || !described.insert(name) {
364            return Err(Error::Metadata {
365                id: stream.id.clone(),
366                reason: format!("field `{name}` is unknown or described more than once"),
367            });
368        }
369    }
370    Ok(())
371}
372
373pub const fn encoded_len(facility: &str, owner: &str, streams: &[Entry]) -> usize {
374    let mut len = HEAD.len() + quoted_len(facility) + OWNER.len() + quoted_len(owner) + STREAMS.len() + TAIL.len();
375    let mut index = 0;
376    while index < streams.len() {
377        len += ID.len()
378            + quoted_len(streams[index].id)
379            + SCHEMA.len()
380            + digits(streams[index].schema as u64)
381            + WINDOW.len()
382            + digits(streams[index].window as u64)
383            + ROW.len()
384            + quoted_len(streams[index].row.name)
385            + METADATA.len()
386            + metadata_len(&streams[index])
387            + ROW_METADATA.len()
388            + streams[index].row_metadata_json.len()
389            + 1;
390        if index > 0 {
391            len += 1;
392        }
393        index += 1;
394    }
395    len
396}
397
398pub const fn encode<const N: usize>(facility: &str, owner: &str, streams: &[Entry]) -> [u8; N] {
399    assert!(N == encoded_len(facility, owner, streams));
400    let mut output = [0; N];
401    let mut cursor = 0;
402    write(&mut output, &mut cursor, HEAD);
403    quoted(&mut output, &mut cursor, facility);
404    write(&mut output, &mut cursor, OWNER);
405    quoted(&mut output, &mut cursor, owner);
406    write(&mut output, &mut cursor, STREAMS);
407    let mut index = 0;
408    while index < streams.len() {
409        if index > 0 {
410            write(&mut output, &mut cursor, b",");
411        }
412        write(&mut output, &mut cursor, ID);
413        quoted(&mut output, &mut cursor, streams[index].id);
414        write(&mut output, &mut cursor, SCHEMA);
415        super::number(&mut output, &mut cursor, streams[index].schema as u64);
416        write(&mut output, &mut cursor, WINDOW);
417        super::number(&mut output, &mut cursor, streams[index].window as u64);
418        write(&mut output, &mut cursor, ROW);
419        quoted(&mut output, &mut cursor, streams[index].row.name);
420        write(&mut output, &mut cursor, METADATA);
421        metadata_into(&mut output, &mut cursor, &streams[index]);
422        write(&mut output, &mut cursor, ROW_METADATA);
423        write(&mut output, &mut cursor, streams[index].row_metadata_json.as_bytes());
424        write(&mut output, &mut cursor, b"}");
425        index += 1;
426    }
427    write(&mut output, &mut cursor, TAIL);
428    output
429}
430
431pub(crate) const fn same_bytes(left: &str, right: &str) -> bool {
432    let left = left.as_bytes();
433    let right = right.as_bytes();
434    if left.len() != right.len() {
435        return false;
436    }
437    let mut i = 0;
438    while i < left.len() {
439        if left[i] != right[i] {
440            return false;
441        }
442        i += 1;
443    }
444    true
445}
446
447/* === Private Functions === */
448
449const fn metadata_len(entry: &Entry) -> usize {
450    if entry.metadata_json.is_empty() {
451        METADATA_ID.len() + quoted_len(entry.id) + METADATA_LABEL.len() + quoted_len(entry.id) + 1
452    } else {
453        entry.metadata_json.len()
454    }
455}
456
457const fn metadata_into<const N: usize>(output: &mut [u8; N], cursor: &mut usize, entry: &Entry) {
458    if entry.metadata_json.is_empty() {
459        write(output, cursor, METADATA_ID);
460        quoted(output, cursor, entry.id);
461        write(output, cursor, METADATA_LABEL);
462        quoted(output, cursor, entry.id);
463        write(output, cursor, b"}");
464    } else {
465        write(output, cursor, entry.metadata_json.as_bytes());
466    }
467}
468
469/* === Tests === */
harmos_domain/inventory/publications.rs

harmos_domain/inventory/
publications.rs
