zeph_db/
pool.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use crate::DbPool;
5use crate::error::DbError;
6
7/// Configuration for database pool construction.
8pub struct DbConfig {
9    /// Database URL. `SQLite`: file path or `:memory:`. `PostgreSQL`: connection URL.
10    pub url: String,
11    /// Maximum number of connections in the pool.
12    pub max_connections: u32,
13    /// `SQLite` only: connection pool size. Default 5.
14    ///
15    /// `BEGIN IMMEDIATE` serializes concurrent writers at the `SQLite` level;
16    /// the pool size controls read concurrency only.
17    pub pool_size: u32,
18}
19
20impl Default for DbConfig {
21    fn default() -> Self {
22        Self {
23            url: String::new(),
24            max_connections: 5,
25            pool_size: 5,
26        }
27    }
28}
29
30impl DbConfig {
31    /// Connect to the database and run migrations.
32    ///
33    /// # Errors
34    ///
35    /// Returns [`DbError`] if connection or migration fails.
36    pub async fn connect(&self) -> Result<DbPool, DbError> {
37        #[cfg(feature = "sqlite")]
38        {
39            Self::connect_sqlite(&self.url, self.max_connections, self.pool_size).await
40        }
41        #[cfg(feature = "postgres")]
42        {
43            Self::connect_postgres(&self.url, self.pool_size).await
44        }
45    }
46
47    #[cfg(feature = "sqlite")]
48    async fn connect_sqlite(
49        path: &str,
50        max_connections: u32,
51        pool_size: u32,
52    ) -> Result<DbPool, DbError> {
53        use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
54        use std::str::FromStr;
55
56        let url = if path == ":memory:" {
57            "sqlite::memory:".to_string()
58        } else {
59            if let Some(parent) = std::path::Path::new(path).parent()
60                && !parent.as_os_str().is_empty()
61            {
62                std::fs::create_dir_all(parent)?;
63            }
64            format!("sqlite:{path}?mode=rwc")
65        };
66
67        let opts = SqliteConnectOptions::from_str(&url)
68            .map_err(DbError::Sqlx)?
69            .create_if_missing(true)
70            .foreign_keys(true)
71            .busy_timeout(std::time::Duration::from_secs(5))
72            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
73            .synchronous(sqlx::sqlite::SqliteSynchronous::Normal);
74
75        // BEGIN IMMEDIATE serializes concurrent writers at the SQLite level.
76        // pool_size controls the connection count; max_connections is the upper bound.
77        let effective_max = max_connections.max(pool_size);
78        let pool = SqlitePoolOptions::new()
79            .max_connections(effective_max)
80            .min_connections(1)
81            .acquire_timeout(std::time::Duration::from_secs(30))
82            .connect_with(opts)
83            .await
84            .map_err(DbError::Sqlx)?;
85
86        crate::migrate::run_migrations(&pool).await?;
87
88        // Restrict file permissions to owner-only on Unix.
89        #[cfg(unix)]
90        if path != ":memory:" {
91            use std::os::unix::fs::PermissionsExt;
92            if let Ok(metadata) = std::fs::metadata(path) {
93                let mut perms = metadata.permissions();
94                perms.set_mode(0o600);
95                let _ = std::fs::set_permissions(path, perms);
96            }
97        }
98
99        // Run a passive WAL checkpoint after migrations to avoid unbounded WAL growth.
100        // Skipped for in-memory databases (no WAL file).
101        if path != ":memory:" {
102            sqlx::query("PRAGMA wal_checkpoint(PASSIVE)")
103                .execute(&pool)
104                .await
105                .map_err(DbError::Sqlx)?;
106        }
107
108        Ok(pool)
109    }
110
111    #[cfg(feature = "postgres")]
112    async fn connect_postgres(url: &str, pool_size: u32) -> Result<DbPool, DbError> {
113        use sqlx::postgres::PgPoolOptions;
114
115        if !url.contains("sslmode=") {
116            tracing::warn!(
117                "postgres connection string has no sslmode; plaintext connections are allowed"
118            );
119        }
120
121        let pool = PgPoolOptions::new()
122            .max_connections(pool_size)
123            .acquire_timeout(std::time::Duration::from_secs(30))
124            .connect(url)
125            .await
126            .map_err(|e| DbError::Connection {
127                url: redact_url(url).unwrap_or_else(|| "[redacted]".into()),
128                source: e,
129            })?;
130
131        crate::migrate::run_migrations(&pool).await?;
132
133        Ok(pool)
134    }
135}
136
137/// Strip password from a database URL for safe logging.
138///
139/// Replaces `://user:password@` with `://[redacted]@`.
140///
141/// Returns `None` if the URL contains no embedded credentials (already safe).
142/// Returns `Some(redacted)` if credentials were found and replaced.
143#[must_use]
144pub fn redact_url(url: &str) -> Option<String> {
145    use std::sync::LazyLock;
146    static RE: LazyLock<regex::Regex> =
147        LazyLock::new(|| regex::Regex::new(r"://[^:]+:[^@]+@").expect("static regex"));
148    if RE.is_match(url) {
149        Some(RE.replace(url, "://[redacted]@").into_owned())
150    } else {
151        None
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    use super::*;
158
159    #[test]
160    fn redact_url_replaces_credentials() {
161        let url = "postgres://user:secret@localhost:5432/zeph";
162        let redacted = redact_url(url).unwrap();
163        assert_eq!(redacted, "postgres://[redacted]@localhost:5432/zeph");
164        assert!(!redacted.contains("secret"));
165    }
166
167    #[test]
168    fn redact_url_returns_none_for_no_credentials() {
169        // URL without credentials — no match, returns None
170        let url = "postgres://localhost:5432/zeph";
171        assert!(redact_url(url).is_none());
172    }
173
174    #[test]
175    fn redact_url_handles_sqlite_path() {
176        let url = "sqlite:/path/to/db";
177        assert!(redact_url(url).is_none());
178    }
179}
zeph_db/pool.rs

zeph_db/
pool.rs
