sofie_db/
contracts.rs

1#[cfg(feature = "jobs")]
2mod jobs;
3#[cfg(feature = "local-time")]
4mod local;
5mod mirror;
6mod provided;
7
8#[cfg(feature = "jobs")]
9pub use jobs::*;
10#[cfg(feature = "local-time")]
11pub(crate) use local::fire_due;
12#[cfg(feature = "local-time")]
13pub use local::{LocalJobs, LocalScheduled, Repeat};
14pub use mirror::{Backfilled, Mirror, MirrorPage, MirrorRow, backfill_mirror, changed_field};
15pub(crate) use mirror::{SWEEP_FORGOTTEN, sweep_forgotten};
16pub(crate) use provided::Provided;
17
18use std::marker::PhantomData;
19use std::ops::Deref;
20use std::sync::Arc;
21
22use serde::Serialize;
23use serde::de::DeserializeOwned;
24use serde_json::{Map, Value};
25use sofie_core::{Error, Result, Uuid};
26
27use crate::{Aggregate, BoxFuture, Db, Id, Resolve, Select, Writes};
28
29pub struct Dep<T: ?Sized>(Arc<T>);
30
31impl<T: ?Sized> Deref for Dep<T> {
32    type Target = T;
33
34    fn deref(&self) -> &T {
35        &self.0
36    }
37}
38
39impl<C: Writes, T: ?Sized + Send + Sync + 'static> Resolve<C> for Dep<T> {
40    fn resolve(context: &mut C) -> impl Future<Output = Result<Self>> + Send {
41        std::future::ready(provided(context.db()).map(Self).ok_or_else(|| {
42            Error::internal_message(format!(
43                "nothing provides {}; run `sofie gen` so the wiring does",
44                std::any::type_name::<T>()
45            ))
46        }))
47    }
48}
49
50pub struct Extensions<T: ?Sized>(Arc<Vec<Arc<T>>>);
51
52impl<T: ?Sized> Deref for Extensions<T> {
53    type Target = [Arc<T>];
54
55    fn deref(&self) -> &[Arc<T>] {
56        &self.0
57    }
58}
59
60impl<C: Writes, T: ?Sized + Send + Sync + 'static> Resolve<C> for Extensions<T> {
61    fn resolve(context: &mut C) -> impl Future<Output = Result<Self>> + Send {
62        let provided = context.db().provided.get::<Vec<Arc<T>>>();
63        std::future::ready(Ok(Self(provided.unwrap_or_default())))
64    }
65}
66
67pub fn provide<T: ?Sized + Send + Sync + 'static>(db: &Db, implementation: Arc<T>) {
68    db.provided
69        .change(|slot: &mut Option<Arc<T>>| *slot = Some(implementation));
70}
71
72/// For a remote contract's server, and a package's arguments that resolve in a GET too.
73#[must_use]
74pub fn provided<T: ?Sized + Send + Sync + 'static>(db: &Db) -> Option<Arc<T>> {
75    db.provided
76        .get::<Option<Arc<T>>>()
77        .and_then(|slot| slot.as_ref().clone())
78}
79
80pub fn extend<T: ?Sized + Send + Sync + 'static>(db: &Db, extension: Arc<T>) {
81    db.provided
82        .change(|extensions: &mut Vec<Arc<T>>| extensions.push(extension));
83}
84
85pub trait Lookup<D>: Send + Sync {
86    fn get_many<'a>(&'a self, ids: &'a [Uuid]) -> BoxFuture<'a, Result<Vec<D>>>;
87
88    /// Records past `after` by id as change events carry them, deleted ones left out.
89    fn changes(
90        &self,
91        after: Option<Uuid>,
92        limit: u64,
93    ) -> BoxFuture<'_, Result<Vec<Map<String, Value>>>> {
94        let _ = (after, limit);
95        Box::pin(std::future::ready(Err(Error::internal_message(
96            "this lookup cannot list its records, so a mirror of it fills only from its change \
97             events",
98        ))))
99    }
100}
101
102pub struct Exposed<A, D> {
103    db: Db,
104    types: PhantomData<fn() -> (A, D)>,
105}
106
107impl<A, D> Exposed<A, D> {
108    #[must_use]
109    pub fn new(db: Db) -> Self {
110        Self {
111            db,
112            types: PhantomData,
113        }
114    }
115}
116
117/// Exposed fields are public to every module, so a lookup reaches every record.
118impl<A: Aggregate + Serialize, D: DeserializeOwned + Send> Lookup<D> for Exposed<A, D> {
119    fn get_many<'a>(&'a self, ids: &'a [Uuid]) -> BoxFuture<'a, Result<Vec<D>>> {
120        Box::pin(async move {
121            let ids = ids.iter().copied().map(Id::from_uuid);
122            let records = A::query()
123                .where_(A::ID.in_(ids))
124                .all(&self.db.read())
125                .await?;
126            records.iter().map(exposed).collect()
127        })
128    }
129
130    fn changes(
131        &self,
132        after: Option<Uuid>,
133        limit: u64,
134    ) -> BoxFuture<'_, Result<Vec<Map<String, Value>>>> {
135        Box::pin(async move {
136            let query = A::query().order_by(A::ID.asc()).limit(limit);
137            let query = match after {
138                Some(after) => query.where_(A::ID.gt(Id::from_uuid(after))),
139                None => query,
140            };
141            let records = query.all(&self.db.read()).await?;
142            records.iter().map(change_of).collect()
143        })
144    }
145}
146
147fn change_of<A: Aggregate>(aggregate: &A) -> Result<Map<String, Value>> {
148    let change = aggregate.exposed_fields().ok_or_else(|| {
149        Error::internal_message(format!("`{}` exposes no fields to change events", A::NAME))
150    })??;
151    let meta = aggregate.meta();
152    Ok(crate::change_event::change_payload(
153        change,
154        meta.id(),
155        meta.version(),
156    ))
157}
158
159pub fn exposed<A: Aggregate + Serialize, D: DeserializeOwned>(aggregate: &A) -> Result<D> {
160    let fields = serde_json::to_value(aggregate)
161        .map_err(|error| Error::internal(format!("serializing a {}", A::NAME), error))?;
162    serde_json::from_value(fields).map_err(|error| {
163        Error::internal(
164            format!("reading the exposed fields of a {}", A::NAME),
165            error,
166        )
167    })
168}
169
170pub fn exposed_fields<A: Aggregate + Serialize>(
171    aggregate: &A,
172    names: &[&str],
173) -> Result<Map<String, Value>> {
174    let Value::Object(mut fields) = serde_json::to_value(aggregate)
175        .map_err(|error| Error::internal(format!("serializing a {}", A::NAME), error))?
176    else {
177        return Err(Error::internal_message(format!(
178            "a {} serializes to no object",
179            A::NAME
180        )));
181    };
182    fields.retain(|name, _| names.contains(&name.as_str()));
183    Ok(fields)
184}
sofie_db/contracts.rs

sofie_db/
contracts.rs
