harmos_operations/host/
catalog.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
17#[cfg(feature = "runtime")]
18use std::{num::NonZeroUsize, sync::Arc};
19
20// Third Party Imports
21#[cfg(feature = "runtime")]
22use serde::Serialize;
23
24// Local Imports
25#[cfg(feature = "runtime")]
26use crate::{Error, Operation, Response};
27use harmos_domain::model::storage::codec::{Bytes, Code, Id, Index, Optional, Text, coded, product};
28use harmos_domain::plugin::Grant;
29use harmos_domain::{Identifier, MemorySize};
30#[cfg(feature = "runtime")]
31use harmos_sdk::operations::access::DiscoveryAccess;
32#[cfg(feature = "runtime")]
33use harmos_storage::{
34    ErasedCatalog, IdentityKind, IdentitySelection, ReadLimits, StorageManager, StorageSchema, StoredIdentity, query::SchemaSelection,
35};
36
37/* === Constants === */
38
39/* === Macros === */
40
41product!(Catalog {
42    kind: Code<CatalogKind> => 1,
43    schema: Optional<Id> => 2,
44    after: Optional<Text> => 3,
45    rows: Optional<Index> => 4,
46    bytes: Optional<Index> => 5,
47});
48
49product!(CatalogPage { entries: Bytes => 1, next: Optional<Text> => 2 });
50
51coded!(CatalogKind { Operations, Schemas, Streams, Objects });
52
53/* === Type Aliases === */
54
55/* === Structs === */
56
57#[harmos::operation(
58    harmos_path = crate,
59    resources = [
60        Option<Arc<StorageManager<ErasedCatalog>>>,
61        DiscoveryAccess,
62    ],
63    codec = manual,
64    id = "host.catalog",
65    response = CatalogPage,
66    policy = super::POLICY,
67    visibility = Shared,
68    grant = Catalog::required_grant,
69    grpc(package = "harmos.host.v1", service = "Host", method = "Catalog"),
70)]
71#[derive(Clone, Debug, PartialEq, Eq)]
72pub struct Catalog {
73    pub kind: CatalogKind,
74    pub schema: Option<Identifier>,
75    pub after: Option<String>,
76    pub rows: Option<usize>,
77    pub bytes: Option<usize>,
78}
79
80#[derive(Clone, Debug, Default, PartialEq, Eq)]
81pub struct CatalogPage {
82    pub entries: Vec<u8>,
83    pub next: Option<String>,
84}
85
86/* === Enums === */
87
88#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
89pub enum CatalogKind {
90    #[default]
91    Operations,
92
93    Schemas,
94    Streams,
95    Objects,
96}
97
98/* === Traits === */
99
100/* === Implementations === */
101
102impl Catalog {
103    // Stored Data Follows The Storage Read Grant Grammar; Operation Discovery
104    // Keeps Its Per-Contract Conditional Access.
105    pub fn required_grant(&self) -> Option<Grant> {
106        let (family, all) = match self.kind {
107            CatalogKind::Operations => return None,
108            CatalogKind::Schemas => ("schemas", "schemas:*"),
109            CatalogKind::Streams => ("streams", "streams:*"),
110            CatalogKind::Objects => ("objects", "objects:*"),
111        };
112        Some(Grant::of(
113            <Catalog as harmos_domain::operations::OperationDeclaration>::ID,
114            self.schema.as_ref().map_or_else(|| all.to_owned(), |schema| format!("{family}:schema:{schema}")),
115        ))
116    }
117}
118
119impl MemorySize for Catalog {
120    fn memory_size(&self) -> usize {
121        std::mem::size_of::<Self>()
122            .saturating_add(self.schema.as_ref().map_or(0, |schema| schema.as_str().len()))
123            .saturating_add(self.after.as_ref().map_or(0, String::capacity))
124    }
125}
126
127impl MemorySize for CatalogPage {
128    fn memory_size(&self) -> usize {
129        std::mem::size_of::<Self>().saturating_add(self.entries.capacity()).saturating_add(self.next.as_ref().map_or(0, String::capacity))
130    }
131}
132
133/* === Public Functions === */
134
135/* === Private Functions === */
136
137/* === Tests === */
138
139#[cfg(feature = "runtime")]
140type Result<T> = std::result::Result<T, Error>;
141#[cfg(feature = "runtime")]
142impl Operation for Catalog {
143    type Error = Error;
144
145    async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<Response<CatalogPage>> {
146        let storage = context.resource::<Option<std::sync::Arc<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>>>();
147        let access = context.resource::<harmos_sdk::operations::access::DiscoveryAccess>();
148
149        let host = access;
150
151        let page = match self.kind {
152            CatalogKind::Operations => {
153                let (rows, bytes) = self.limits()?;
154                let (entries, next) = host.0.discover(self.after.as_deref(), rows, bytes).map_err(Error::Access)?;
155                CatalogPage { entries, next }
156            }
157            CatalogKind::Schemas => {
158                let schema = self.schema.clone();
159                stored::<StorageSchema>(
160                    storage.as_deref().as_ref().ok_or(Error::NotReady { domain: "catalog storage" })?,
161                    &SchemaSelection { schema: None },
162                    &self,
163                    |entry| schema.as_ref().is_none_or(|schema| schema == &entry.key.id),
164                )
165                .await?
166            }
167            CatalogKind::Streams | CatalogKind::Objects => {
168                let selection = IdentitySelection {
169                    kind: if self.kind == CatalogKind::Streams { IdentityKind::Streams } else { IdentityKind::Objects },
170                    schema: self.schema.clone(),
171                };
172                stored::<StoredIdentity>(
173                    storage.as_deref().as_ref().ok_or(Error::NotReady { domain: "catalog storage" })?,
174                    &selection,
175                    &self,
176                    |_| true,
177                )
178                .await?
179            }
180        };
181        Ok(page.into())
182    }
183}
184
185#[cfg(feature = "runtime")]
186async fn stored<T>(
187    storage: &StorageManager<ErasedCatalog>, selection: &T::Selection, request: &Catalog, keep: impl Fn(&T) -> bool,
188) -> Result<CatalogPage>
189where
190    T: harmos_storage::follow::Entry,
191{
192    let (rows, bytes) = request.limits()?;
193    let after = request
194        .after
195        .as_deref()
196        .map(str::parse::<u64>)
197        .transpose()
198        .map_err(|_| Error::InvalidInput {
199            field: "catalog cursor",
200            reason: "cursor must be a storage position",
201        })?
202        .unwrap_or(0);
203    let limits = ReadLimits {
204        rows: NonZeroUsize::new(rows).expect("validated catalog rows"),
205        bytes: NonZeroUsize::new(bytes).expect("validated catalog bytes"),
206    };
207    let page = storage.page::<T>(selection, after, limits).await?;
208    let mut entries = vec![b'['];
209    let mut next = None;
210    let mut count = 0usize;
211    for entry in page.entries() {
212        let position = entry.position().sequence;
213        if keep(entry) {
214            let start = entries.len();
215            if count > 0 {
216                entries.push(b',');
217            }
218            if write(&mut entries, entry).is_err() || entries.len() + 1 > bytes {
219                entries.truncate(start);
220                if next.is_none() {
221                    return Err(Error::ResourceExhausted { resource: "catalog entry bytes" });
222                }
223                break;
224            }
225            count += 1;
226        }
227        next = Some(position.to_string());
228    }
229    entries.push(b']');
230    Ok(CatalogPage { entries, next })
231}
232
233#[cfg(feature = "runtime")]
234fn write(bytes: &mut Vec<u8>, entry: &impl Serialize) -> std::result::Result<(), serde_json::Error> {
235    serde_json::to_writer(bytes, entry)
236}
237
238impl Catalog {
239    pub fn limits(&self) -> std::result::Result<(usize, usize), harmos_domain::MessageError> {
240        let rows = self.rows.unwrap_or(64);
241        let bytes = self.bytes.unwrap_or(64 * 1024);
242        if rows == 0 || rows > 256 || !(2..=4 * 1024 * 1024).contains(&bytes) {
243            return Err(harmos_domain::MessageError::borrowed("rows must be 1..=256 and bytes must be 2..=4194304"));
244        }
245        Ok((rows, bytes))
246    }
247}
harmos_operations/host/catalog.rs

harmos_operations/host/
catalog.rs
