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:V6:{\"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)
158                    .map_err(|source| Error::Fields { id: stream.id.clone(), source: source.into() })?;
159                let row =
160                    super::retain_schema(row, settings).map_err(|source| Error::Fields { id: stream.id.clone(), source: source.into() })?;
161                let stream = StreamDeclaration {
162                    metadata: stream.metadata,
163                    row_metadata: stream.row_metadata,
164                    id: stream.id,
165                    schema: stream.schema,
166                    window: stream.window,
167                    row,
168                };
169                validate(&stream)?;
170                Ok(stream)
171            })
172            .collect::<Result<_, Error>>()?;
173        Ok(Manifest { facility: self.facility, owner: self.owner, streams })
174    }
175}
176
177impl Entry {
178    pub fn metadata(&self) -> crate::Metadata {
179        if self.metadata_json.is_empty() {
180            crate::Metadata::new(crate::Identifier::declared(self.id, "stream"), self.id, None)
181        } else {
182            serde_json::from_str(self.metadata_json).expect("declared stream metadata is valid")
183        }
184    }
185
186    pub fn row_metadata(&self) -> Vec<crate::StreamField> {
187        serde_json::from_str(self.row_metadata_json).expect("declared row metadata is valid")
188    }
189
190    pub const fn of<S: StreamSpec>() -> Self
191    where
192        S::Row: crate::schema::Described,
193    {
194        Self {
195            id: S::ID,
196            metadata_json: S::METADATA_JSON,
197            row_metadata_json: S::ROW_METADATA_JSON,
198            schema: S::VERSION,
199            window: S::WINDOW,
200            row: <S::Row as crate::schema::Described>::DEFINITION,
201        }
202    }
203}
204
205/* === Public Functions === */
206
207pub const fn merge<const N: usize>(named: &[Entry], sampled: &[Entry]) -> [Entry; N] {
208    let mut merged = [Entry { id: "", metadata_json: "", row_metadata_json: "[]", schema: 0, window: 0, row: Definition::of::<()>() }; N];
209    let mut index = 0;
210    while index < named.len() {
211        merged[index] = named[index];
212        index += 1;
213    }
214    let mut offset = 0;
215    while offset < sampled.len() {
216        merged[index + offset] = sampled[offset];
217        offset += 1;
218    }
219
220    merged
221}
222
223pub const fn require_tree_subset(observers: &[crate::catalog::Node<super::observers::Entry>], grants: &[Entry]) {
224    let mut index = 0;
225    while index < observers.len() {
226        match &observers[index] {
227            crate::catalog::Node::Leaf(entry) => require_subset(&[entry.streams], grants),
228            crate::catalog::Node::Nested(nested) => require_tree_subset(nested, grants),
229        }
230        index += 1;
231    }
232}
233
234pub const fn require_subset(outputs: &[&[Entry]], grants: &[Entry]) {
235    let mut observer = 0;
236    while observer < outputs.len() {
237        let mut output = 0;
238        while output < outputs[observer].len() {
239            let required = &outputs[observer][output];
240            let mut grant = 0;
241            let mut found = false;
242            while grant < grants.len() {
243                let declared = &grants[grant];
244                if same_bytes(required.id, declared.id) && required.schema == declared.schema && required.window == declared.window {
245                    found = true;
246                    break;
247                }
248                grant += 1;
249            }
250            assert!(found, "observer outputs must be a subset of the artifact publication grant");
251            output += 1;
252        }
253        observer += 1;
254    }
255}
256
257pub fn select(manifests: Vec<Manifest>, facility: &str, owner: &str) -> Result<Vec<StreamDeclaration>, Error> {
258    let mut matching = manifests.into_iter().filter(|m| m.facility == facility && m.owner == owner);
259    let manifest = matching.next().ok_or_else(|| Error::Missing { facility: facility.into(), owner: owner.into() })?;
260    if matching.next().is_some() {
261        return Err(Error::Duplicate { facility: facility.into(), owner: owner.into() });
262    }
263    let mut streams = BTreeMap::new();
264    for stream in manifest.streams {
265        validate(&stream)?;
266        let id = stream.id.clone();
267        if streams.insert(id.clone(), stream).is_some() {
268            return Err(Error::DuplicateStream { id });
269        }
270    }
271    Ok(streams.into_values().collect())
272}
273
274pub fn validate_match(expected: &[Entry], actual: &[StreamDeclaration]) -> Result<(), Error> {
275    let mut authored = BTreeMap::new();
276    for entry in expected {
277        let metadata = if entry.metadata_json.is_empty() {
278            let id = crate::Identifier::new(entry.id).map_err(|_| Error::InvalidId { id: entry.id.to_owned() })?;
279            crate::Metadata::new(id, entry.id, None)
280        } else {
281            serde_json::from_str(entry.metadata_json)
282                .map_err(|source| Error::Metadata { id: entry.id.to_owned(), reason: source.to_string() })?
283        };
284        let row_metadata = serde_json::from_str(entry.row_metadata_json)
285            .map_err(|source| Error::Metadata { id: entry.id.to_owned(), reason: source.to_string() })?;
286        let stream = StreamDeclaration {
287            metadata,
288            row_metadata,
289            id: entry.id.to_owned(),
290            schema: entry.schema,
291            window: entry.window,
292            row: Schema::from_declaration(entry.row, &SchemaSettings::default())
293                .map_err(|source| Error::Fields { id: entry.id.to_owned(), source: source.into() })?,
294        };
295        validate(&stream)?;
296        if authored.insert(entry.id, stream).is_some() {
297            return Err(Error::DuplicateStream { id: entry.id.to_owned() });
298        }
299    }
300    let mut admitted = BTreeMap::new();
301    for stream in actual {
302        validate(stream)?;
303        if admitted.insert(stream.id.as_str(), stream).is_some() {
304            return Err(Error::DuplicateStream { id: stream.id.clone() });
305        }
306    }
307    if authored.len() != admitted.len() || authored.iter().any(|(id, stream)| admitted.get(id).is_none_or(|actual| *actual != stream)) {
308        return Err(Error::Mismatch);
309    }
310    Ok(())
311}
312
313pub fn validate(stream: &StreamDeclaration) -> Result<(), Error> {
314    if !crate::stream_subset(&stream.id) {
315        return Err(Error::InvalidId { id: stream.id.clone() });
316    }
317    if stream.schema == 0 {
318        return Err(Error::ZeroSchema { id: stream.id.clone() });
319    }
320    if stream.window == 0 {
321        return Err(Error::ZeroWindow { id: stream.id.clone() });
322    }
323    if stream.metadata.identifier.as_str() != stream.id {
324        return Err(Error::Metadata { id: stream.id.clone(), reason: "identifier differs from stream identity".into() });
325    }
326    stream.row.prepare(&SchemaSettings::default()).map_err(|source| Error::Fields { id: stream.id.clone(), source: source.into() })?;
327    let mut described = std::collections::BTreeSet::new();
328    for field in &stream.row_metadata {
329        let name = field.metadata.identifier.as_str();
330        if !stream.row.definitions[stream.row.root].iter().any(|wire| wire.name == name) || !described.insert(name) {
331            return Err(Error::Metadata {
332                id: stream.id.clone(),
333                reason: format!("field `{name}` is unknown or described more than once"),
334            });
335        }
336    }
337    Ok(())
338}
339
340pub const fn encoded_len(facility: &str, owner: &str, streams: &[Entry]) -> usize {
341    let mut len = HEAD.len() + quoted_len(facility) + OWNER.len() + quoted_len(owner) + STREAMS.len() + TAIL.len();
342    let mut index = 0;
343    while index < streams.len() {
344        len += ID.len()
345            + quoted_len(streams[index].id)
346            + SCHEMA.len()
347            + digits(streams[index].schema as u64)
348            + WINDOW.len()
349            + digits(streams[index].window as u64)
350            + ROW.len()
351            + quoted_len(streams[index].row.name)
352            + METADATA.len()
353            + metadata_len(&streams[index])
354            + ROW_METADATA.len()
355            + streams[index].row_metadata_json.len()
356            + 1;
357        if index > 0 {
358            len += 1;
359        }
360        index += 1;
361    }
362    len
363}
364
365pub const fn encode<const N: usize>(facility: &str, owner: &str, streams: &[Entry]) -> [u8; N] {
366    assert!(N == encoded_len(facility, owner, streams));
367    let mut output = [0; N];
368    let mut cursor = 0;
369    write(&mut output, &mut cursor, HEAD);
370    quoted(&mut output, &mut cursor, facility);
371    write(&mut output, &mut cursor, OWNER);
372    quoted(&mut output, &mut cursor, owner);
373    write(&mut output, &mut cursor, STREAMS);
374    let mut index = 0;
375    while index < streams.len() {
376        if index > 0 {
377            write(&mut output, &mut cursor, b",");
378        }
379        write(&mut output, &mut cursor, ID);
380        quoted(&mut output, &mut cursor, streams[index].id);
381        write(&mut output, &mut cursor, SCHEMA);
382        super::number(&mut output, &mut cursor, streams[index].schema as u64);
383        write(&mut output, &mut cursor, WINDOW);
384        super::number(&mut output, &mut cursor, streams[index].window as u64);
385        write(&mut output, &mut cursor, ROW);
386        quoted(&mut output, &mut cursor, streams[index].row.name);
387        write(&mut output, &mut cursor, METADATA);
388        metadata_into(&mut output, &mut cursor, &streams[index]);
389        write(&mut output, &mut cursor, ROW_METADATA);
390        write(&mut output, &mut cursor, streams[index].row_metadata_json.as_bytes());
391        write(&mut output, &mut cursor, b"}");
392        index += 1;
393    }
394    write(&mut output, &mut cursor, TAIL);
395    output
396}
397
398pub(crate) const fn same_bytes(left: &str, right: &str) -> bool {
399    let left = left.as_bytes();
400    let right = right.as_bytes();
401    if left.len() != right.len() {
402        return false;
403    }
404    let mut i = 0;
405    while i < left.len() {
406        if left[i] != right[i] {
407            return false;
408        }
409        i += 1;
410    }
411    true
412}
413
414/* === Private Functions === */
415
416const fn metadata_len(entry: &Entry) -> usize {
417    if entry.metadata_json.is_empty() {
418        METADATA_ID.len() + quoted_len(entry.id) + METADATA_LABEL.len() + quoted_len(entry.id) + 1
419    } else {
420        entry.metadata_json.len()
421    }
422}
423
424const fn metadata_into<const N: usize>(output: &mut [u8; N], cursor: &mut usize, entry: &Entry) {
425    if entry.metadata_json.is_empty() {
426        write(output, cursor, METADATA_ID);
427        quoted(output, cursor, entry.id);
428        write(output, cursor, METADATA_LABEL);
429        quoted(output, cursor, entry.id);
430        write(output, cursor, b"}");
431    } else {
432        write(output, cursor, entry.metadata_json.as_bytes());
433    }
434}
435
436/* === Tests === */
harmos_domain/inventory/publications.rs

harmos_domain/inventory/
publications.rs
