bhtune_db/

pool.rs

//! Database connection setup: WAL journal mode, a busy timeout (so concurrent CLI + GUI
//! access to the same file doesn't immediately error out with `SQLITE_BUSY`), and
//! foreign-key enforcement (off by default in SQLite, and required here since every child
//! table relies on `ON DELETE CASCADE`/`ON DELETE SET NULL`).

use std::{
    path::{Path, PathBuf},
    time::Duration,
};

use sqlx::{
    Row, SqlitePool,
    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions},
};

use crate::error::{DbError, DbResult};

/// How long a connection waits on a locked database before giving up (`SQLITE_BUSY`).
/// Generous enough to ride out a retention sweep or a large history export running
/// concurrently in another process, without silently hanging forever if something is
/// actually deadlocked.
const BUSY_TIMEOUT: Duration = Duration::from_secs(10);

static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");

/// Opens (creating if missing) the SQLite database at `path`, applies the standard pragmas,
/// rejects incompatible migration histories, and runs pending migrations. This is
/// the only supported way to get a `SqlitePool` in bhtune — every caller (CLI, web server,
/// tests) goes through this function so the pragmas and migration state can never drift
/// between adapters.
pub async fn connect(path: &Path) -> DbResult<SqlitePool> {
    let options = SqliteConnectOptions::new()
        .filename(path)
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal)
        .busy_timeout(BUSY_TIMEOUT)
        .foreign_keys(true);

    let pool = SqlitePoolOptions::new()
        .connect_with(options)
        .await
        .map_err(DbError::Connect)?;

    run_migrations(&pool, path).await?;

    Ok(pool)
}

/// Opens an existing SQLite database without creating it or applying migrations.
///
/// Incompatible migration history is rejected before returning the pool. This is for commands
/// that need a snapshot of persisted settings but must not mutate the database as a side effect.
/// Callers must not use the returned pool for writes; SQLite opens it in read-only mode and
/// rejects them.
pub async fn connect_read_only(path: &Path) -> DbResult<SqlitePool> {
    let options = SqliteConnectOptions::new()
        .filename(path)
        .read_only(true)
        .busy_timeout(BUSY_TIMEOUT)
        .foreign_keys(true);

    let pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect_with(options)
        .await
        .map_err(DbError::Connect)?;

    if let Err(error) = validate_migration_history(&pool, path).await {
        pool.close().await;
        return Err(error);
    }

    Ok(pool)
}

/// Returns the filesystem path attached to SQLite's `main` database, or
/// [`DbError::DatabasePathUnavailable`] for private in-memory databases. Live controller
/// mutations require this stable identity to acquire process-shared ownership.
/// Callers canonicalize the returned path before comparing filesystem aliases.
pub async fn database_path(pool: &SqlitePool) -> DbResult<PathBuf> {
    let rows = sqlx::query("PRAGMA database_list")
        .fetch_all(pool)
        .await
        .map_err(DbError::Query)?;
    let entries = rows
        .into_iter()
        .map(|row| {
            Ok((
                row.try_get("name").map_err(DbError::Query)?,
                row.try_get("file").map_err(DbError::Query)?,
            ))
        })
        .collect::<DbResult<Vec<_>>>()?;
    database_path_from_entries(entries)
}

fn database_path_from_entries(
    entries: impl IntoIterator<Item = (String, String)>,
) -> DbResult<PathBuf> {
    for (name, path) in entries {
        if name == "main" {
            if path.is_empty() {
                return Err(DbError::DatabasePathUnavailable);
            }
            return Ok(PathBuf::from(path));
        }
    }
    Err(DbError::DatabasePathUnavailable)
}

/// Opens a private, in-process database for tests: same pragmas and migrations as
/// [`connect`], but nothing touches disk.
///
/// Capped at one pooled connection deliberately — a pooled in-memory SQLite database is
/// private per-connection, so a second connection would see an *empty* database rather than
/// the same one. That cap is what makes "in-memory" behave like a single shared database
/// across a test. WAL itself requires a real file (SQLite silently falls back to its
/// in-memory journal mode here regardless of the request below), so WAL is only actually
/// exercised by [`connect`]'s own tests, which use a real temp file.
pub async fn connect_in_memory() -> DbResult<SqlitePool> {
    let options = SqliteConnectOptions::new()
        .filename(":memory:")
        .journal_mode(SqliteJournalMode::Wal)
        .busy_timeout(BUSY_TIMEOUT)
        .foreign_keys(true);

    let pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect_with(options)
        .await
        .map_err(DbError::Connect)?;

    run_migrations(&pool, Path::new(":memory:")).await?;

    Ok(pool)
}

async fn run_migrations(pool: &SqlitePool, path: &Path) -> DbResult<()> {
    validate_migration_history(pool, path).await?;
    MIGRATOR.run(pool).await.map_err(DbError::Migrate)
}

pub(crate) async fn validate_migration_history(pool: &SqlitePool, path: &Path) -> DbResult<()> {
    let has_migration_table: i64 = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = '_sqlx_migrations')",
    )
    .fetch_one(pool)
    .await
    .map_err(DbError::Query)?;
    if has_migration_table == 0 {
        return Ok(());
    }

    for row in sqlx::query("SELECT version, checksum FROM _sqlx_migrations ORDER BY version")
        .fetch_all(pool)
        .await
        .map_err(DbError::Query)?
    {
        let version: i64 = row.try_get("version").map_err(DbError::Query)?;
        let checksum: Vec<u8> = row.try_get("checksum").map_err(DbError::Query)?;
        if !MIGRATOR.iter().any(|migration| {
            migration.version == version && migration.checksum.as_ref() == checksum.as_slice()
        }) {
            return Err(DbError::IncompatibleMigrationHistory(path.to_path_buf()));
        }
    }

    Ok(())
}

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

    #[tokio::test]
    async fn connect_creates_file_and_applies_wal_and_foreign_keys() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("bhtune.db");

        let pool = connect(&path).await.unwrap();
        assert!(path.exists());

        let (journal_mode,): (String,) = sqlx::query_as("PRAGMA journal_mode")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(journal_mode, "wal");

        let (foreign_keys,): (i64,) = sqlx::query_as("PRAGMA foreign_keys")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(foreign_keys, 1);
    }

    #[tokio::test]
    async fn connect_is_idempotent_across_reopens() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("bhtune.db");

        connect(&path).await.unwrap();
        // Reopening an already-migrated database (the normal "app restart" case) must not
        // fail or try to reapply migrations.
        let pool = connect(&path).await.unwrap();

        let (count,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'tune_runs'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(count, 1);

        let (migration_count,): (i64,) =
            sqlx::query_as("SELECT COUNT(*) FROM _sqlx_migrations WHERE success = 1")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(migration_count, 1);
    }

    #[tokio::test]
    async fn connect_read_only_does_not_create_a_missing_database() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("missing.db");

        assert!(connect_read_only(&path).await.is_err());
        assert!(!path.exists());
    }

    #[tokio::test]
    async fn database_path_returns_the_main_database_file() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("bhtune.db");
        let pool = connect(&path).await.unwrap();

        assert_eq!(
            std::fs::canonicalize(database_path(&pool).await.unwrap()).unwrap(),
            std::fs::canonicalize(path).unwrap()
        );
    }

    #[tokio::test]
    async fn database_path_rejects_in_memory_databases() {
        let pool = connect_in_memory().await.unwrap();

        assert!(matches!(
            database_path(&pool).await,
            Err(DbError::DatabasePathUnavailable)
        ));
    }

    #[test]
    fn database_path_rejects_a_listing_without_the_main_database() {
        assert!(matches!(
            database_path_from_entries([("temp".to_string(), String::new())]),
            Err(DbError::DatabasePathUnavailable)
        ));
    }

    #[tokio::test]
    async fn connect_read_only_queries_existing_database_and_rejects_writes() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("bhtune.db");
        let pool = connect(&path).await.unwrap();
        drop(pool);

        let pool = connect_read_only(&path).await.unwrap();
        let migration_count: i64 =
            sqlx::query_scalar("SELECT COUNT(*) FROM _sqlx_migrations WHERE success = 1")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(migration_count, 1);

        assert!(
            sqlx::query("DELETE FROM tune_runs")
                .execute(&pool)
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn connect_in_memory_runs_the_final_schema_migration() {
        let pool = connect_in_memory().await.unwrap();

        let (migration_count,): (i64,) =
            sqlx::query_as("SELECT COUNT(*) FROM _sqlx_migrations WHERE success = 1")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(migration_count, 1);

        let (migration_version,): (i64,) = sqlx::query_as(
            "SELECT version FROM _sqlx_migrations WHERE success = 1 ORDER BY version DESC LIMIT 1",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(migration_version, 1);
    }

    #[tokio::test]
    async fn fresh_schema_contains_final_history_demo_and_actuation_objects() {
        let pool = connect_in_memory().await.unwrap();

        for (object_type, object_name) in [
            ("table", "demo_sessions"),
            ("table", "tune_mv_actuations"),
            ("table", "live_operation_owners"),
            ("table", "live_operation_claims"),
            ("table", "live_mutation_steps"),
            ("table", "tune_recovery_attempts"),
            ("index", "idx_tune_samples_run_time"),
            ("index", "idx_tune_writes_run_written"),
            ("index", "idx_tune_runs_demo_session"),
            ("index", "idx_tune_runs_demo_session_outcome"),
            ("index", "idx_live_operation_owners_run"),
            ("index", "idx_live_operation_owners_heartbeat"),
            ("index", "idx_live_mutation_steps_owner"),
            ("index", "idx_live_mutation_steps_run"),
            ("index", "idx_tune_recovery_attempts_run"),
            ("trigger", "tune_runs_demo_session_insert"),
            ("trigger", "tune_runs_demo_session_valid_insert"),
            ("trigger", "tune_runs_demo_global_limit_insert"),
            ("trigger", "tune_runs_demo_session_update"),
            ("trigger", "tune_runs_demo_session_immutable"),
        ] {
            let count: i64 = sqlx::query_scalar(
                "SELECT COUNT(*) FROM sqlite_master WHERE type = ? AND name = ?",
            )
            .bind(object_type)
            .bind(object_name)
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                count, 1,
                "expected final schema object {object_type} {object_name}"
            );
        }

        let result_columns: Vec<String> =
            sqlx::query_scalar("SELECT name FROM pragma_table_info('tune_results')")
                .fetch_all(&pool)
                .await
                .unwrap();
        for column in ["status", "invalid_reason"] {
            assert!(
                result_columns.iter().any(|name| name == column),
                "checked-result column {column} is missing"
            );
        }
    }
}