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

harmos_domain/inventory/
publications.rs
