oxide_batch_cli/
backend.rs

1//! The `PostgreSQL` repository backend.
2//!
3//! This module is the only place a resolved [`Secret`] is exposed, and it
4//! exposes it solely to construct a connection. A connection string, host name,
5//! user name, or certificate never leaves this module.
6
7use std::sync::Arc;
8
9use oxide_batch::{
10    BoxFuture, CaCertificate, Clock, JobExplorer, JobOperator, OwnerToken, PostgresConfig,
11    PostgresExplorer, PostgresJobRepository, PostgresMigrator, RecoveryProposer, RepositoryError,
12    RetentionService, SystemClock, SystemMonotonicClock, TlsMode,
13};
14
15use crate::config::{Configuration, TlsSetting};
16use crate::exit::ExitCategory;
17use crate::output::Diagnostic;
18use crate::run::{SchemaReport, SchemaState, Services};
19
20/// The services one `PostgreSQL` deployment provides.
21pub type PostgresServices = Services<PostgresJobRepository, PostgresExplorer>;
22
23/// A failure to open the repository.
24#[derive(Clone, Debug, Eq, PartialEq)]
25pub struct BackendFailure {
26    category: ExitCategory,
27    diagnostic: Diagnostic,
28}
29
30impl BackendFailure {
31    /// Returns the exit category this failure reports.
32    #[must_use]
33    pub const fn category(&self) -> ExitCategory {
34        self.category
35    }
36
37    /// Borrows the redacted diagnostic.
38    #[must_use]
39    pub const fn diagnostic(&self) -> &Diagnostic {
40        &self.diagnostic
41    }
42}
43
44/// Reports the durable schema version of one `PostgreSQL` deployment.
45struct PostgresSchema {
46    config: PostgresConfig,
47}
48
49impl SchemaReport for PostgresSchema {
50    fn schema_state(&self) -> BoxFuture<'_, Result<SchemaState, RepositoryError>> {
51        Box::pin(async move {
52            let installed = PostgresMigrator::installed_schema_version(&self.config).await?;
53            Ok(SchemaState {
54                installed,
55                supported: PostgresMigrator::supported_schema_version(),
56            })
57        })
58    }
59}
60
61/// Builds the connection configuration from resolved settings.
62///
63/// # Errors
64///
65/// Returns a redacted failure when no connection string was supplied or when a
66/// bounded value is outside the adapter's accepted range.
67pub fn connection_config(config: &Configuration) -> Result<PostgresConfig, BackendFailure> {
68    let url = config.repository_url().ok_or_else(|| BackendFailure {
69        category: ExitCategory::ConfigurationInvalid,
70        diagnostic: Diagnostic::new(
71            "REPOSITORY_URL_MISSING",
72            "no repository connection was configured; set OXIDE_BATCH_REPOSITORY_URL \
73             or repository.url",
74        ),
75    })?;
76    let invalid = |detail: &'static str| BackendFailure {
77        category: ExitCategory::ConfigurationInvalid,
78        diagnostic: Diagnostic::new("REPOSITORY_CONFIG_INVALID", detail),
79    };
80    let mut built = PostgresConfig::new(url.value().expose())
81        .map_err(|_| invalid("the repository connection string is not accepted"))?;
82    built = built
83        .with_pool_size(config.pool_size())
84        .map_err(|_| invalid("the repository pool size is outside its accepted range"))?
85        .with_connect_timeout(config.connect_timeout())
86        .map_err(|_| invalid("the connect timeout is outside its accepted range"))?
87        .with_statement_timeout(config.statement_timeout())
88        .map_err(|_| invalid("the statement timeout is outside its accepted range"))?;
89    let tls = match config.tls_mode() {
90        TlsSetting::Plaintext => TlsMode::Plaintext,
91        TlsSetting::VerifyFull => {
92            let ca_certificate = match config.ca_certificate() {
93                None => None,
94                Some(pem) => Some(
95                    CaCertificate::new(pem.value().expose().as_bytes().to_vec())
96                        .map_err(|_| invalid("the certificate authority bundle is not accepted"))?,
97                ),
98            };
99            TlsMode::VerifyFull { ca_certificate }
100        }
101    };
102    Ok(built.with_tls_mode(tls))
103}
104
105/// Opens the repository and binds the portable services.
106///
107/// # Errors
108///
109/// Returns a redacted failure when the configuration is not accepted or the
110/// repository is unavailable.
111pub async fn connect(config: &Configuration) -> Result<PostgresServices, BackendFailure> {
112    let connection = connection_config(config)?;
113    let clock: Arc<dyn Clock> = Arc::new(SystemClock);
114    let repository = PostgresJobRepository::connect(connection.clone(), clock.clone())
115        .await
116        .map_err(|error| BackendFailure {
117            category: crate::failure::repository(&error),
118            diagnostic: crate::failure::repository_diagnostic(&error),
119        })?;
120    let explorer_repository = PostgresExplorer::new(repository.clone());
121    let recovery = RecoveryProposer::new(
122        explorer_repository.clone(),
123        Arc::clone(&clock),
124        Arc::new(SystemMonotonicClock::new()),
125        OwnerToken::from_bytes([0; 16]),
126    );
127    let explorer = JobExplorer::new(explorer_repository);
128    let operator = JobOperator::new(repository.clone(), clock.clone());
129    let retention = RetentionService::new(repository, clock);
130    Ok(Services::new(
131        operator,
132        retention,
133        explorer,
134        Box::new(PostgresSchema { config: connection }),
135    )
136    .with_recovery_proposals(Box::new(recovery)))
137}
oxide_batch_cli/backend.rs

oxide_batch_cli/
backend.rs
