harmos_runtime/operations/
listing.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// The Paged Catalog Read: Live Operation Contracts From The Manager, And The
11// Schemas, Streams And Objects Storage Holds Durably From Its Saved Frontier.
12
13/* === Modules === */
14
15/* === Re-Exports === */
16
17/* === Imports === */
18
19// Standard Library Imports
20use std::num::NonZeroUsize;
21use std::sync::Arc;
22
23// Third Party Imports
24use serde::Serialize;
25
26// Local Imports
27use super::host::{Catalog, CatalogKind, CatalogPage};
28use super::{Context, Lane, Operation, OperationManager};
29use crate::storage::query::SchemaSelection;
30use crate::storage::{IdentityKind, IdentitySelection, ReadLimits, StorageManager, StorageSchema, StoredIdentity};
31use crate::{Error, Result};
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39/* === Structs === */
40
41pub struct Listing<C> {
42    storage: Arc<Option<Arc<StorageManager<C>>>>,
43}
44
45/* === Enums === */
46
47/* === Traits === */
48
49/* === Implementations === */
50
51impl<C: 'static> Operation for Listing<C> {
52    type Declaration = Catalog;
53    type Resources = Option<Arc<StorageManager<C>>>;
54    type Error = Error;
55    type Context<'a> = Context<'a>;
56    type Endpoints = ();
57
58    async fn initialize<'ctx>(storage: Arc<Self::Resources>, _: Self::Context<'ctx>) -> Result<Self> {
59        Ok(Self { storage })
60    }
61
62    async fn run<'ctx>(
63        &mut self, context: Self::Context<'ctx>, request: &mut Catalog, _: &mut Self::Endpoints,
64    ) -> Result<super::Response<CatalogPage>> {
65        let page = match request.kind {
66            CatalogKind::Operations => context.operations.discover(&context.source, request)?,
67            CatalogKind::Schemas => {
68                let schema = request.schema.clone();
69                stored::<C, StorageSchema>(self.storage()?, &SchemaSelection { schema: None }, request, |entry| {
70                    schema.as_ref().is_none_or(|schema| schema == &entry.key.id)
71                })
72                .await?
73            }
74            CatalogKind::Streams | CatalogKind::Objects => {
75                let selection = IdentitySelection {
76                    kind: if request.kind == CatalogKind::Streams { IdentityKind::Streams } else { IdentityKind::Objects },
77                    schema: request.schema.clone(),
78                };
79                stored::<C, StoredIdentity>(self.storage()?, &selection, request, |_| true).await?
80            }
81        };
82        Ok(page.into())
83    }
84}
85
86impl<C> Listing<C> {
87    fn storage(&self) -> Result<&StorageManager<C>> {
88        self.storage.as_deref().ok_or(Error::NotReady { domain: "catalog storage" })
89    }
90}
91
92/* === Public Functions === */
93
94pub fn register<C: 'static>(
95    operations: &OperationManager, owner: impl Into<String>, lane: impl Lane, storage: Option<Arc<StorageManager<C>>>,
96) -> Result<()> {
97    operations.register_wire_operation::<Listing<C>>(owner.into(), lane, Arc::new(storage))
98}
99
100/* === Private Functions === */
101
102// One Page Of Durable Entries After The Request's Storage Position. Pages Order
103// By Storage Position, So A Nonempty Page Always Continues And An Empty Page
104// Ends The Catalog.
105async fn stored<C, T>(
106    storage: &StorageManager<C>, selection: &T::Selection, request: &Catalog, keep: impl Fn(&T) -> bool,
107) -> Result<CatalogPage>
108where
109    T: crate::storage::follow::Entry,
110{
111    let (rows, bytes) = super::manager::limits(request)?;
112    let after = request
113        .after
114        .as_deref()
115        .map(str::parse::<u64>)
116        .transpose()
117        .map_err(|_| Error::InvalidInput { field: "catalog cursor", reason: "cursor must be a storage position" })?
118        .unwrap_or(0);
119    let limits = ReadLimits {
120        rows: NonZeroUsize::new(rows).expect("validated catalog rows"),
121        bytes: NonZeroUsize::new(bytes).expect("validated catalog bytes"),
122    };
123    let page = storage.page::<T>(selection, after, limits).await?;
124    let mut entries = vec![b'['];
125    let mut next = None;
126    let mut count = 0usize;
127    for entry in page.entries() {
128        let position = entry.position().sequence;
129        if keep(entry) {
130            let start = entries.len();
131            if count > 0 {
132                entries.push(b',');
133            }
134            if write(&mut entries, entry).is_err() || entries.len() + 1 > bytes {
135                entries.truncate(start);
136                if next.is_none() {
137                    return Err(Error::ResourceExhausted { resource: "catalog entry bytes" });
138                }
139                break;
140            }
141            count += 1;
142        }
143        next = Some(position.to_string());
144    }
145    entries.push(b']');
146    Ok(CatalogPage { entries, next })
147}
148
149fn write(bytes: &mut Vec<u8>, entry: &impl Serialize) -> std::result::Result<(), serde_json::Error> {
150    serde_json::to_writer(bytes, entry)
151}
152
153/* === Tests === */
harmos_runtime/operations/listing.rs

harmos_runtime/operations/
listing.rs
