zeph_db/

pool.rs

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

use crate::DbPool;
use crate::error::DbError;

/// Configuration for database pool construction.
pub struct DbConfig {
    /// Database URL. `SQLite`: file path or `:memory:`. `PostgreSQL`: connection URL.
    pub url: String,
    /// Maximum number of connections in the pool.
    pub max_connections: u32,
    /// `SQLite` only: connection pool size. Default 5.
    ///
    /// `BEGIN IMMEDIATE` serializes concurrent writers at the `SQLite` level;
    /// the pool size controls read concurrency only.
    pub pool_size: u32,
}

impl Default for DbConfig {
    fn default() -> Self {
        Self {
            url: String::new(),
            max_connections: 5,
            pool_size: 5,
        }
    }
}

impl DbConfig {
    /// Connect to the database and run migrations.
    ///
    /// # Errors
    ///
    /// Returns [`DbError`] if connection or migration fails.
    #[tracing::instrument(name = "db.pool.connect", skip_all, err)]
    pub async fn connect(&self) -> Result<DbPool, DbError> {
        #[cfg(all(feature = "sqlite", not(feature = "postgres")))]
        {
            Self::connect_sqlite(&self.url, self.max_connections, self.pool_size).await
        }
        #[cfg(feature = "postgres")]
        {
            Self::connect_postgres(&self.url, self.pool_size).await
        }
    }

    #[cfg(all(feature = "sqlite", not(feature = "postgres")))]
    async fn connect_sqlite(
        path: &str,
        max_connections: u32,
        pool_size: u32,
    ) -> Result<DbPool, DbError> {
        use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
        use std::str::FromStr;

        let url = if path == ":memory:" {
            "sqlite::memory:".to_string()
        } else {
            let db_path = std::path::PathBuf::from(path);

            if let Some(parent) = db_path.parent()
                && !parent.as_os_str().is_empty()
            {
                tokio::fs::create_dir_all(parent).await?;
            }
            // Pre-create with 0o600 so sqlx inherits the mode rather than using the
            // process umask. sqlx reopens the existing file via SQLITE_OPEN_CREATE.
            // WAL/SHM sidecars are created by sqlx after the pool opens and will still
            // inherit the process umask (sqlx limitation — best-effort chmod below).
            if tokio::fs::metadata(&db_path).await.is_err() {
                let p = db_path.clone();
                tokio::task::spawn_blocking(move || {
                    zeph_common::fs_secure::open_private_truncate(&p)
                })
                .await
                .map_err(|e| std::io::Error::other(format!("spawn_blocking panicked: {e}")))??;
            }
            format!("sqlite:{path}?mode=rwc")
        };

        let opts = SqliteConnectOptions::from_str(&url)
            .map_err(DbError::Sqlx)?
            .create_if_missing(true)
            .foreign_keys(true)
            .busy_timeout(std::time::Duration::from_secs(5))
            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
            .synchronous(sqlx::sqlite::SqliteSynchronous::Normal);

        // BEGIN IMMEDIATE serializes concurrent writers at the SQLite level.
        // pool_size controls the connection count; max_connections is the upper bound.
        // In-memory databases are connection-scoped: each new connection is a separate
        // empty DB. Force a single connection so all queries share the migrated schema.
        let effective_max = if path == ":memory:" {
            1
        } else {
            max_connections.max(pool_size)
        };
        let pool = SqlitePoolOptions::new()
            .max_connections(effective_max)
            .min_connections(1)
            .acquire_timeout(std::time::Duration::from_secs(30))
            .connect_with(opts)
            .await
            .map_err(DbError::Sqlx)?;

        crate::migrate::run_migrations(&pool).await?;

        // Best-effort chmod for .db, .db-wal, and .db-shm. The .db itself was
        // pre-created with 0o600 above; the WAL/SHM sidecars are created by sqlx
        // after the pool opens and inherit the process umask, so we fix them here.
        // There is a small race window between sidecar creation and this chmod;
        // there is no way to close it without upstream sqlx support.
        #[cfg(unix)]
        if path != ":memory:" {
            let path_owned = path.to_owned();
            tokio::task::spawn_blocking(move || {
                use std::os::unix::fs::PermissionsExt as _;
                for suffix in &["", "-wal", "-shm", "-journal"] {
                    let p = format!("{path_owned}{suffix}");
                    if let Ok(metadata) = std::fs::metadata(&p) {
                        let mut perms = metadata.permissions();
                        perms.set_mode(0o600);
                        let _ = std::fs::set_permissions(&p, perms);
                    }
                }
            })
            .await
            .map_err(|e| std::io::Error::other(format!("spawn_blocking panicked: {e}")))?;
        }

        // Run a passive WAL checkpoint after migrations to avoid unbounded WAL growth.
        // Skipped for in-memory databases (no WAL file).
        if path != ":memory:" {
            sqlx::query("PRAGMA wal_checkpoint(PASSIVE)")
                .execute(&pool)
                .await
                .map_err(DbError::Sqlx)?;
        }

        Ok(pool)
    }

    #[cfg(feature = "postgres")]
    async fn connect_postgres(url: &str, pool_size: u32) -> Result<DbPool, DbError> {
        use sqlx::postgres::PgPoolOptions;

        if !url.contains("sslmode=") {
            tracing::warn!(
                "postgres connection string has no sslmode; plaintext connections are allowed"
            );
        }

        let pool = PgPoolOptions::new()
            .max_connections(pool_size)
            .acquire_timeout(std::time::Duration::from_secs(30))
            .connect(url)
            .await
            .map_err(|e| DbError::Connection {
                url: redact_url(url).unwrap_or_else(|| "[redacted]".into()),
                source: e,
            })?;

        crate::migrate::run_migrations(&pool).await?;

        Ok(pool)
    }
}

/// Strip password from a database URL for safe logging.
///
/// Replaces `://user:password@` with `://[redacted]@`.
///
/// Returns `None` if the URL contains no embedded credentials (already safe).
/// Returns `Some(redacted)` if credentials were found and replaced.
#[must_use]
pub fn redact_url(url: &str) -> Option<String> {
    use std::sync::LazyLock;
    static RE: LazyLock<regex::Regex> =
        LazyLock::new(|| regex::Regex::new(r"://[^:]+:[^@]+@").expect("static regex"));
    if RE.is_match(url) {
        Some(RE.replace(url, "://[redacted]@").into_owned())
    } else {
        None
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn redact_url_replaces_credentials() {
        let url = "postgres://user:secret@localhost:5432/zeph";
        let redacted = redact_url(url).unwrap();
        assert_eq!(redacted, "postgres://[redacted]@localhost:5432/zeph");
        assert!(!redacted.contains("secret"));
    }

    #[test]
    fn redact_url_returns_none_for_no_credentials() {
        // URL without credentials — no match, returns None
        let url = "postgres://localhost:5432/zeph";
        assert!(redact_url(url).is_none());
    }

    #[test]
    fn redact_url_handles_sqlite_path() {
        let url = "sqlite:/path/to/db";
        assert!(redact_url(url).is_none());
    }

    #[cfg(all(unix, feature = "sqlite", not(feature = "postgres")))]
    #[tokio::test]
    async fn sqlite_precreated_with_0600() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let cfg = DbConfig {
            url: db_path.to_str().unwrap().to_owned(),
            max_connections: 1,
            pool_size: 1,
        };
        cfg.connect().await.unwrap();
        let mode = std::fs::metadata(&db_path).unwrap().permissions().mode() & 0o777;
        assert_eq!(
            mode, 0o600,
            "SQLite DB file must be created with mode 0o600"
        );
    }
}