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    let mut matching = manifests.into_iter().filter(|manifest| manifest.facility == facility && manifest.owner == owner);
167    let manifest = matching.next().ok_or_else(|| Error::Missing { facility: facility.into(), owner: owner.into() })?;
168    if matching.next().is_some() {
169        return Err(Error::Duplicate { facility: facility.into(), owner: owner.into() });
170    }
171
172    validate(&manifest.observers, grants)?;
173    Ok(manifest)
174}
175
176pub fn validate(observers: &[Declaration], grants: &[StreamDeclaration]) -> Result<(), Error> {
177    let mut ids = BTreeSet::new();
178    for observer in observers {
179        let invalid = |reason| Error::Invalid { id: observer.id.clone(), reason };
180        if observer.id.trim().is_empty() || observer.id.chars().any(char::is_control) || !ids.insert(&observer.id) {
181            return Err(invalid("identities must be nonempty, unique, and contain no control characters"));
182        }
183        if observer.interval.is_zero() {
184            return Err(invalid("interval must be positive"));
185        }
186        let mut keys = BTreeSet::new();
187        for stream in &observer.streams {
188            if !keys.insert(stream) || !grants.iter().any(|grant| grant.id == *stream) {
189                return Err(invalid("outputs must be unique members of the artifact publication grant"));
190            }
191        }
192        for key in &observer.demand_keys {
193            if key.trim().is_empty() || key.chars().any(char::is_control) || !keys.insert(key) {
194                return Err(invalid("demand keys must be nonempty, unique, and distinct from outputs"));
195            }
196        }
197        if observer.activation == Activation::Demand && keys.is_empty() {
198            return Err(invalid("demand activation requires an output or demand key"));
199        }
200    }
201    Ok(())
202}
203
204pub const fn encoded_len(facility: &str, owner: &str, entries: &[Entry]) -> usize {
205    encoded::<0>(facility, owner, entries).cursor
206}
207
208pub const fn encode<const N: usize>(facility: &str, owner: &str, entries: &[Entry]) -> [u8; N] {
209    assert!(N == encoded_len(facility, owner, entries));
210    encoded::<N>(facility, owner, entries).bytes
211}
212
213pub const fn encoded_tree_len(facility: &str, owner: &str, entries: &[Node<Entry>]) -> usize {
214    let length = encoded_tree::<0>(facility, owner, entries).cursor;
215    assert!(length - crate::PREFIX.len() - 1 <= crate::FRAGMENT_LIMIT);
216    length
217}
218
219pub const fn encode_tree<const N: usize>(facility: &str, owner: &str, entries: &[Node<Entry>]) -> [u8; N] {
220    assert!(N == encoded_tree_len(facility, owner, entries));
221    encoded_tree::<N>(facility, owner, entries).bytes
222}
223
224/* === Private Functions === */
225
226const fn encoded<const N: usize>(facility: &str, owner: &str, entries: &[Entry]) -> Encoder<N> {
227    let mut out = Encoder { bytes: [0; N], cursor: 0 };
228    out.raw(b"\0HARMOS:ARTIFACT:V6:{\"kind\":\"observers\",\"facility\":");
229    out.quoted(facility);
230    out.raw(b",\"owner\":");
231    out.quoted(owner);
232    out.raw(b",\"observers\":[");
233    let mut index = 0;
234    while index < entries.len() {
235        if index > 0 {
236            out.raw(b",");
237        }
238        entry(&mut out, entries[index]);
239        index += 1;
240    }
241    out.raw(b"]}\0");
242    out
243}
244
245const fn entry<const N: usize>(out: &mut Encoder<N>, entry: Entry) {
246    out.raw(b"{\"id\":");
247    out.quoted(entry.id);
248    out.raw(b",\"description\":");
249    out.quoted(entry.description);
250    out.raw(b",\"interval\":{\"secs\":");
251    out.number(entry.interval.as_secs());
252    out.raw(b",\"nanos\":");
253    out.number(entry.interval.subsec_nanos() as u64);
254    out.raw(b"},\"activation\":");
255    out.quoted(match entry.activation {
256        Activation::Plugin => "plugin",
257        Activation::Demand => "demand",
258    });
259    out.raw(b",\"streams\":[");
260    let mut item = 0;
261    while item < entry.streams.len() {
262        if item > 0 {
263            out.raw(b",");
264        }
265        out.quoted(entry.streams[item].id);
266        item += 1;
267    }
268    out.raw(b"],\"demand_keys\":[");
269    item = 0;
270    while item < entry.demand_keys.len() {
271        if item > 0 {
272            out.raw(b",");
273        }
274        out.key(entry.demand_keys[item]);
275        item += 1;
276    }
277    out.raw(b"]}");
278}
279
280const fn encoded_tree<const N: usize>(facility: &str, owner: &str, entries: &[Node<Entry>]) -> Encoder<N> {
281    let mut out = Encoder { bytes: [0; N], cursor: 0 };
282    out.raw(b"\0HARMOS:ARTIFACT:V6:{\"kind\":\"observers\",\"facility\":");
283    out.quoted(facility);
284    out.raw(b",\"owner\":");
285    out.quoted(owner);
286    out.raw(b",\"observers\":[");
287    let mut first = true;
288    tree(&mut out, entries, &mut first, 0);
289    out.raw(b"]");
290    out.raw(b"}\0");
291    out
292}
293
294const fn tree<const N: usize>(out: &mut Encoder<N>, entries: &[Node<Entry>], first: &mut bool, depth: usize) {
295    assert!(depth <= 32);
296    let mut index = 0;
297    while index < entries.len() {
298        match &entries[index] {
299            Node::Leaf(value) => {
300                if !*first {
301                    out.raw(b",");
302                }
303                *first = false;
304                entry(out, *value);
305            }
306            Node::Nested(entries) => tree(out, entries, first, depth + 1),
307        }
308        index += 1;
309    }
310}
311
312/* === Tests === */
harmos_domain/inventory/observers.rs

harmos_domain/inventory/
observers.rs
