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

sofie_db/
db.rs
