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 { domain: "native record schema", name: schema.id.as_str().to_owned() });
66        }
67        if !pending.is_committed()
68            && let Some(storage) = &self.schemas.storage
69        {
70            storage.write_schema(pending.definition.schema().clone(), pending.definition.graph_owner()).await?;
71        }
72        Ok(pending.commit())
73    }
74
75    pub fn definition(&self, schema: &SchemaKey) -> Option<Arc<Definition>> {
76        self.schemas.definition(schema)
77    }
78
79    pub fn definitions(&self) -> Vec<Arc<Definition>> {
80        self.schemas.definitions()
81    }
82
83    pub async fn publish<R: Into<A::Record>>(&self, record: R) -> Result<Receipt> {
84        let record = record.into();
85        let preparation =
86            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
87        let prepared = Prepared::typed_record(&record).map_err(Error::Preparation)?;
88        let declared = self.catalog.has_record(&prepared.id, prepared.version).map_err(|source| Error::RecordDeclaration { source })?;
89        if !declared {
90            return Err(Error::RecordDeclaration { source: crate::state::DecodeError::UnknownDefinition { id: prepared.id } });
91        }
92        self.state.append_prepared(Recorded::Record(record), prepared, preparation).await
93    }
94
95    pub async fn publish_system(&self, fact: SystemFact) -> Result<Receipt> {
96        let preparation =
97            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
98        let prepared = Prepared::system(&fact).map_err(Error::Preparation)?;
99        self.state.append_prepared(Recorded::System(fact), prepared, preparation).await
100    }
101
102    pub fn as_principal(&self, principal: PrincipalId) -> Self {
103        Self::new(self.state.as_principal(principal))
104    }
105
106    pub fn caused_by(&self, cause: Position) -> Self {
107        Self::new(self.state.caused_by(cause))
108    }
109}
110
111impl<A: Application> Clone for RecordManager<A> {
112    fn clone(&self) -> Self {
113        Self::new(self.state.clone())
114    }
115}
116
117/* === Public Functions === */
118
119/* === Private Functions === */
120
121/* === Tests === */
harmos_runtime/records/manager.rs

harmos_runtime/records/
manager.rs
