harmos_domain/inventory/
observers.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::BTreeSet;
18use std::time::Duration;
19// Third Party Imports
20use serde::{Deserialize, Serialize};
21// Local Imports
22use super::contracts::{self, Encoder};
23use super::{StreamDeclaration, publications};
24use crate::catalog::Node;
25
26/* === Constants === */
27
28/* === Macros === */
29
30/* === Type Aliases === */
31
32/* === Structs === */
33
34pub struct Descriptors;
35
36#[derive(Clone, Copy, Debug, PartialEq, Eq)]
37pub struct DemandKey {
38    namespace: &'static str,
39    id: &'static str,
40}
41
42#[derive(Clone, Copy)]
43pub struct Entry {
44    pub id: &'static str,
45    pub description: &'static str,
46    pub interval: Duration,
47    pub activation: Activation,
48    pub streams: &'static [publications::Entry],
49    pub demand_keys: &'static [DemandKey],
50}
51
52#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
53#[serde(deny_unknown_fields)]
54pub struct Declaration {
55    pub id: String,
56    pub description: String,
57    pub interval: Duration,
58    pub activation: Activation,
59    pub streams: Vec<String>,
60    pub demand_keys: Vec<String>,
61}
62
63#[derive(Deserialize)]
64#[serde(deny_unknown_fields)]
65pub struct Manifest {
66    pub facility: String,
67    pub owner: String,
68    pub observers: Vec<Declaration>,
69}
70
71/* === Enums === */
72
73#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
74#[serde(rename_all = "snake_case")]
75pub enum Activation {
76    #[default]
77    Plugin,
78
79    Demand,
80}
81
82#[derive(Debug, thiserror::Error)]
83pub enum Error {
84    #[error(transparent)]
85    Contract(#[from] contracts::Error),
86
87    #[error(transparent)]
88    Schema(#[from] crate::schema::graph::Error),
89
90    #[error("the artifact carries no complete {facility} observer inventory for `{owner}`")]
91    Missing { facility: String, owner: String },
92
93    #[error("the artifact carries multiple {facility} observer inventories for `{owner}`")]
94    Duplicate { facility: String, owner: String },
95
96    #[error("observer `{id}` is invalid: {reason}")]
97    Invalid { id: String, reason: &'static str },
98}
99
100/* === Traits === */
101
102pub trait Declared {
103    const ENTRY: Entry;
104}
105
106/* === Implementations === */
107
108impl crate::catalog::Visitor for Descriptors {
109    type Entry = Entry;
110}
111
112impl<S: Declared> crate::catalog::Member<S> for Descriptors {
113    const ENTRY: Entry = S::ENTRY;
114}
115
116impl Entry {
117    pub fn declaration(&self) -> Declaration {
118        Declaration {
119            id: self.id.to_owned(),
120            description: self.description.to_owned(),
121            interval: self.interval,
122            activation: self.activation,
123            streams: self.streams.iter().map(|entry| entry.id.to_owned()).collect(),
124            demand_keys: self.demand_keys.iter().map(DemandKey::name).collect(),
125        }
126    }
127}
128
129impl DemandKey {
130    pub const fn new(namespace: &'static str, id: &'static str) -> Self {
131        assert!(!namespace.is_empty() && !id.is_empty(), "demand keys require a namespace and identity");
132        Self { namespace, id }
133    }
134
135    pub fn name(&self) -> String {
136        format!("{}:{}", self.namespace, self.id)
137    }
138}
139
140impl<const N: usize> Encoder<N> {
141    const fn key(&mut self, key: DemandKey) {
142        // Encode both components with the ordinary JSON escaping, sharing quotes.
143        self.quoted(key.namespace);
144        self.cursor -= 1;
145        self.raw(b":");
146        let start = self.cursor;
147        self.quoted(key.id);
148        if N > 0 {
149            let mut index = start;
150            while index + 1 < self.cursor {
151                self.bytes[index] = self.bytes[index + 1];
152                index += 1;
153            }
154        }
155        self.cursor -= 1;
156    }
157}
158
159/* === Public Functions === */
160
161pub fn select(manifests: Vec<Manifest>, facility: &str, owner: &str, grants: &[StreamDeclaration]) -> Result<Vec<Declaration>, Error> {
162    Ok(select_manifest(manifests, facility, owner, grants)?.observers)
163}
164
165pub fn select_manifest(manifests: Vec<Manifest>, facility: &str, owner: &str, grants: &[StreamDeclaration]) -> Result<Manifest, Error> {
166    if !contracts::valid_identity(facility, owner) {
167        return Err(contracts::Error::Identity.into());
168    }
169    let mut matching = manifests.into_iter().filter(|manifest| manifest.facility == facility && manifest.owner == owner);
170    let manifest = matching.next().ok_or_else(|| Error::Missing {
171        facility: facility.into(),
172        owner: owner.into(),
173    })?;
174    if matching.next().is_some() {
175        return Err(Error::Duplicate {
176            facility: facility.into(),
177            owner: owner.into(),
178        });
179    }
180
181    validate(&manifest.observers, grants)?;
182    Ok(manifest)
183}
184
185pub fn validate(observers: &[Declaration], grants: &[StreamDeclaration]) -> Result<(), Error> {
186    let mut ids = BTreeSet::new();
187    for observer in observers {
188        let invalid = |reason| Error::Invalid { id: observer.id.clone(), reason };
189        if observer.id.trim().is_empty() || observer.id.chars().any(char::is_control) || !ids.insert(&observer.id) {
190            return Err(invalid("identities must be nonempty, unique, and contain no control characters"));
191        }
192        if observer.interval.is_zero() {
193            return Err(invalid("interval must be positive"));
194        }
195        let mut keys = BTreeSet::new();
196        for stream in &observer.streams {
197            if !keys.insert(stream) || !grants.iter().any(|grant| grant.id == *stream) {
198                return Err(invalid("outputs must be unique members of the artifact publication grant"));
199            }
200        }
201        for key in &observer.demand_keys {
202            if key.trim().is_empty() || key.chars().any(char::is_control) || !keys.insert(key) {
203                return Err(invalid("demand keys must be nonempty, unique, and distinct from outputs"));
204            }
205        }
206        if observer.activation == Activation::Demand && keys.is_empty() {
207            return Err(invalid("demand activation requires an output or demand key"));
208        }
209    }
210    Ok(())
211}
212
213pub const fn encoded_len(facility: &str, owner: &str, entries: &[Entry]) -> usize {
214    encoded::<0>(facility, owner, entries).cursor
215}
216
217pub const fn encode<const N: usize>(facility: &str, owner: &str, entries: &[Entry]) -> [u8; N] {
218    assert!(N == encoded_len(facility, owner, entries));
219    encoded::<N>(facility, owner, entries).bytes
220}
221
222pub const fn encoded_tree_len(facility: &str, owner: &str, entries: &[Node<Entry>]) -> usize {
223    let length = encoded_tree::<0>(facility, owner, entries).cursor;
224    assert!(length - crate::PREFIX.len() - 1 <= crate::FRAGMENT_LIMIT);
225    length
226}
227
228pub const fn encode_tree<const N: usize>(facility: &str, owner: &str, entries: &[Node<Entry>]) -> [u8; N] {
229    assert!(N == encoded_tree_len(facility, owner, entries));
230    encoded_tree::<N>(facility, owner, entries).bytes
231}
232
233/* === Private Functions === */
234
235const fn encoded<const N: usize>(facility: &str, owner: &str, entries: &[Entry]) -> Encoder<N> {
236    assert!(contracts::valid_identity(facility, owner), "invalid observer inventory identity");
237    let mut out = Encoder { bytes: [0; N], cursor: 0 };
238    out.raw(b"\0HARMOS:ARTIFACT:V1:{\"kind\":\"observers\",\"facility\":");
239    out.quoted(facility);
240    out.raw(b",\"owner\":");
241    out.quoted(owner);
242    out.raw(b",\"observers\":[");
243    let mut index = 0;
244    while index < entries.len() {
245        if index > 0 {
246            out.raw(b",");
247        }
248        entry(&mut out, entries[index]);
249        index += 1;
250    }
251    out.raw(b"]}\0");
252    out
253}
254
255const fn entry<const N: usize>(out: &mut Encoder<N>, entry: Entry) {
256    out.raw(b"{\"id\":");
257    out.quoted(entry.id);
258    out.raw(b",\"description\":");
259    out.quoted(entry.description);
260    out.raw(b",\"interval\":{\"secs\":");
261    out.number(entry.interval.as_secs());
262    out.raw(b",\"nanos\":");
263    out.number(entry.interval.subsec_nanos() as u64);
264    out.raw(b"},\"activation\":");
265    out.quoted(match entry.activation {
266        Activation::Plugin => "plugin",
267        Activation::Demand => "demand",
268    });
269    out.raw(b",\"streams\":[");
270    let mut item = 0;
271    while item < entry.streams.len() {
272        if item > 0 {
273            out.raw(b",");
274        }
275        out.quoted(entry.streams[item].id);
276        item += 1;
277    }
278    out.raw(b"],\"demand_keys\":[");
279    item = 0;
280    while item < entry.demand_keys.len() {
281        if item > 0 {
282            out.raw(b",");
283        }
284        out.key(entry.demand_keys[item]);
285        item += 1;
286    }
287    out.raw(b"]}");
288}
289
290const fn encoded_tree<const N: usize>(facility: &str, owner: &str, entries: &[Node<Entry>]) -> Encoder<N> {
291    assert!(contracts::valid_identity(facility, owner), "invalid observer inventory identity");
292    let mut out = Encoder { bytes: [0; N], cursor: 0 };
293    out.raw(b"\0HARMOS:ARTIFACT:V1:{\"kind\":\"observers\",\"facility\":");
294    out.quoted(facility);
295    out.raw(b",\"owner\":");
296    out.quoted(owner);
297    out.raw(b",\"observers\":[");
298    let mut first = true;
299    tree(&mut out, entries, &mut first, 0);
300    out.raw(b"]");
301    out.raw(b"}\0");
302    out
303}
304
305const fn tree<const N: usize>(out: &mut Encoder<N>, entries: &[Node<Entry>], first: &mut bool, depth: usize) {
306    assert!(depth <= 32);
307    let mut index = 0;
308    while index < entries.len() {
309        match &entries[index] {
310            Node::Leaf(value) => {
311                if !*first {
312                    out.raw(b",");
313                }
314                *first = false;
315                entry(out, *value);
316            }
317            Node::Nested(entries) => tree(out, entries, first, depth + 1),
318        }
319        index += 1;
320    }
321}
322
323/* === Tests === */
harmos_domain/inventory/observers.rs

harmos_domain/inventory/
observers.rs
