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

harmos_domain/inventory/
observers.rs
