harmos_runtime/records/
manager.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::sync::Arc;
18// Third Party Imports
19use tokio::sync::Semaphore;
20// Local Imports
21use super::{Definition, Registry};
22use crate::state::dispatch::Dispatch;
23use crate::storage::SchemaKey;
24use crate::{
25    Application, Error, Result, state::Position, state::Prepared, state::PrincipalId, state::Receipt, state::Recorded, state::StateManager,
26    state::SystemFact,
27};
28
29/* === Constants === */
30
31/* === Macros === */
32
33/* === Type Aliases === */
34
35/* === Structs === */
36
37pub struct RecordManager<A: Application> {
38    pub(super) state: StateManager<A>,
39    pub(super) catalog: Arc<Dispatch<A, crate::storage::Payload<A>>>,
40    preparing: Arc<Semaphore>,
41    pub(super) schemas: Arc<Registry>,
42}
43
44/* === Enums === */
45
46/* === Traits === */
47
48/* === Implementations === */
49
50impl<A: Application> RecordManager<A> {
51    pub fn new(state: StateManager<A>) -> Self {
52        let (catalog, preparing) = state.record_resources();
53        let schemas = state.record_schemas();
54        Self { state, catalog, preparing, schemas }
55    }
56
57    pub async fn register(
58        &self, owner: &str, schema: SchemaKey, graph: impl Into<Arc<crate::schema::graph::Schema>>,
59    ) -> Result<Arc<Definition>> {
60        let _preparation =
61            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
62        let pending = self.schemas.reserve(owner, schema.clone(), graph.into())?;
63        let id = crate::DefinitionId::new(schema.id.as_str());
64        if !matches!(self.catalog.has_record(&id, schema.version.get()), Ok(false)) {
65            return Err(Error::AlreadyRegistered {
66                domain: "native record schema",
67                name: schema.id.as_str().to_owned(),
68            });
69        }
70        if !pending.is_committed()
71            && let Some(storage) = &self.schemas.storage
72        {
73            storage.write_schema(pending.definition.schema().clone(), pending.definition.graph_owner()).await?;
74        }
75        Ok(pending.commit())
76    }
77
78    pub fn definition(&self, schema: &SchemaKey) -> Option<Arc<Definition>> {
79        self.schemas.definition(schema)
80    }
81
82    pub fn definitions(&self) -> Vec<Arc<Definition>> {
83        self.schemas.definitions()
84    }
85
86    pub async fn publish<R: Into<A::Record>>(&self, record: R) -> Result<Receipt> {
87        let record = record.into();
88        let preparation =
89            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
90        let prepared = Prepared::typed_record(&record).map_err(Error::Preparation)?;
91        let declared = self.catalog.has_record(&prepared.id, prepared.version).map_err(|source| Error::RecordDeclaration { source })?;
92        if !declared {
93            return Err(Error::RecordDeclaration {
94                source: crate::state::DecodeError::UnknownDefinition { id: prepared.id },
95            });
96        }
97        self.state.append_prepared(Recorded::Record(record), prepared, preparation).await
98    }
99
100    pub async fn publish_system(&self, fact: SystemFact) -> Result<Receipt> {
101        let preparation =
102            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
103        let prepared = Prepared::system(&fact).map_err(Error::Preparation)?;
104        self.state.append_prepared(Recorded::System(fact), prepared, preparation).await
105    }
106
107    pub fn as_principal(&self, principal: PrincipalId) -> Self {
108        Self::new(self.state.as_principal(principal))
109    }
110
111    pub fn caused_by(&self, cause: Position) -> Self {
112        Self::new(self.state.caused_by(cause))
113    }
114}
115
116impl<A: Application> Clone for RecordManager<A> {
117    fn clone(&self) -> Self {
118        Self::new(self.state.clone())
119    }
120}
121
122/* === Public Functions === */
123
124/* === Private Functions === */
125
126/* === Tests === */
harmos_runtime/records/manager.rs

harmos_runtime/records/
manager.rs
