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

harmos_runtime/records/
manager.rs
