sofie_db/
db.rs

1use std::sync::Arc;
2
3use sofie_auth::SigningKeys;
4use sofie_core::{Clock, Result};
5use tracing::Instrument;
6
7use crate::arrivals::Arrivals;
8use crate::outcome::explain_duplicate;
9use crate::{Driver, Outcome, Plan, Reach, Read, Schema, Step};
10
11#[derive(Clone)]
12pub struct Db {
13    pub(crate) driver: Arc<dyn Driver>,
14    clock: Clock,
15    arrivals: Arc<Arrivals>,
16    keys: Option<Arc<SigningKeys>>,
17    #[cfg(feature = "contracts")]
18    pub(crate) provided: Arc<crate::contracts::Provided>,
19    #[cfg(feature = "cache")]
20    pub(crate) commit_hooks: Arc<crate::commit_hooks::CommitHooks>,
21    #[cfg(feature = "live")]
22    pub(crate) changes: Arc<crate::live::Changes>,
23    #[cfg(feature = "search-engine")]
24    pub(crate) search: Arc<std::sync::OnceLock<crate::search::Installed>>,
25}
26
27impl Db {
28    pub fn new(driver: impl Driver) -> Self {
29        #[cfg(feature = "metrics")]
30        let driver = crate::metrics::Counted::new(driver);
31        let driver: Arc<dyn Driver> = Arc::new(driver);
32        Self {
33            #[cfg(feature = "live")]
34            changes: Arc::new(crate::live::Changes::new(&driver)),
35            driver,
36            clock: Clock::system(),
37            arrivals: Arc::default(),
38            keys: None,
39            #[cfg(feature = "contracts")]
40            provided: Arc::default(),
41            #[cfg(feature = "cache")]
42            commit_hooks: Arc::default(),
43            #[cfg(feature = "search-engine")]
44            search: Arc::default(),
45        }
46    }
47
48    #[must_use]
49    pub fn with_clock(self, clock: Clock) -> Self {
50        Self { clock, ..self }
51    }
52
53    /// Wakes, hooks and screens of `other` then see this one's commits as their own.
54    #[must_use]
55    pub fn sharing_arrivals(self, other: &Db) -> Self {
56        Self {
57            arrivals: Arc::clone(&other.arrivals),
58            #[cfg(feature = "cache")]
59            commit_hooks: Arc::clone(&other.commit_hooks),
60            #[cfg(feature = "live")]
61            changes: Arc::clone(&other.changes),
62            ..self
63        }
64    }
65
66    #[must_use]
67    pub fn with_signing_keys(self, keys: Arc<SigningKeys>) -> Self {
68        Self {
69            keys: Some(keys),
70            ..self
71        }
72    }
73
74    pub(crate) fn signing_keys(&self) -> Option<Arc<SigningKeys>> {
75        self.keys.clone()
76    }
77
78    #[must_use]
79    pub fn clock(&self) -> Clock {
80        self.clock.clone()
81    }
82
83    #[must_use]
84    pub fn read(&self) -> Read {
85        Read::within(self, Reach::All)
86    }
87
88    #[must_use]
89    pub fn schema(&self) -> Schema<'_> {
90        Schema { db: self }
91    }
92
93    pub(crate) fn arrivals(&self) -> &Arrivals {
94        &self.arrivals
95    }
96
97    pub(crate) fn driver(&self) -> &dyn Driver {
98        self.driver.as_ref()
99    }
100
101    /// Opens no transaction when there is nothing to write.
102    pub async fn commit(&self, outcome: &mut dyn Outcome) -> Result<()> {
103        self.commit_with(outcome, Vec::new(), Reach::All).await
104    }
105
106    /// `steps` run first; nothing is written when it would reach a record beyond `reach`.
107    pub async fn commit_with(
108        &self,
109        outcome: &mut dyn Outcome,
110        steps: Vec<Step>,
111        reach: Reach,
112    ) -> Result<()> {
113        let now = self.clock.now();
114        let heard = self.arrivals.heard().known();
115        let mut plan = Plan::new(self.driver().dialect(), reach, now, steps, heard);
116        #[cfg(feature = "search-engine")]
117        if self.search.get().is_some() {
118            plan.index_searches();
119        }
120        outcome.plan(&mut plan)?;
121        let stamp = plan.stamp();
122        let raised = self.arrivals.copy_raised(&plan.events);
123        let planned = plan.into_parts()?;
124        if !planned.steps.is_empty() {
125            let committed = self.transaction(planned.steps).await;
126            planned.guard_checks.iter().for_each(|check| check.audit());
127            committed.map_err(|error| explain_duplicate(error, &planned.unique_fields))?;
128        }
129        let queued = planned.queues_messages;
130        self.arrivals.committed(&self.driver, queued, raised);
131        #[cfg(feature = "cache")]
132        self.commit_hooks.run(&planned.changed);
133        #[cfg(feature = "live")]
134        self.changes.announce(&self.driver, planned.live);
135        outcome.committed(stamp);
136        Ok(())
137    }
138
139    pub async fn ping(&self) -> Result<()> {
140        crate::query_values(self.driver(), "SELECT 1", &[])
141            .await
142            .map(drop)
143    }
144
145    pub(crate) async fn transaction(&self, steps: Vec<Step>) -> Result<()> {
146        let span = tracing::info_span!("commit", db.statements = steps.len());
147        self.driver.transaction(steps).instrument(span).await
148    }
149}
sofie_db/db.rs

sofie_db/
db.rs
