shared-context-engineering 0.3.0-pre-alpha-v5

//! Shared Turso database infrastructure.
//!
//! Provides a generic `TursoDb` adapter that wraps Turso connection
//! management, tokio runtime bridging, and embedded migration execution for
//! service-specific database specs.

use std::{
    fs,
    marker::PhantomData,
    path::{Path, PathBuf},
};

use anyhow::{Context, Result};
use turso::Value as TursoValue;

use crate::services::lifecycle::{
    HealthCategory, HealthFixability, HealthProblem, HealthProblemKind, HealthSeverity,
};
use crate::services::resilience::{run_with_retry_sync, RetryPolicy};

const MIGRATIONS_TABLE_SQL: &str = "CREATE TABLE IF NOT EXISTS __sce_migrations (
    id TEXT PRIMARY KEY,
    applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
)";
const SELECT_MIGRATION_SQL: &str = "SELECT id FROM __sce_migrations WHERE id = ?1 LIMIT 1";
const INSERT_MIGRATION_SQL: &str = "INSERT INTO __sce_migrations (id) VALUES (?1)";
const ENCRYPTION_CIPHER_AEGIS256: &str = "aegis256";
const CONNECTION_OPEN_RETRY_POLICY: RetryPolicy = RetryPolicy {
    max_attempts: 3,
    timeout_ms: 1_000,
    initial_backoff_ms: 25,
    max_backoff_ms: 200,
};
const CONNECTION_OPEN_RETRY_HINT: &str = "retry after the database lock clears; if the issue persists, stop other SCE processes using this database and rerun the command";
const QUERY_RETRY_POLICY: RetryPolicy = RetryPolicy {
    max_attempts: 5,
    timeout_ms: 200,
    initial_backoff_ms: 25,
    max_backoff_ms: 100,
};
const QUERY_RETRY_HINT: &str = "retry after the database lock clears; if the issue persists, stop other SCE processes using this database and rerun the command";

pub mod encryption_key;

/// Service-specific Turso database configuration.
pub trait DbSpec {
    /// Human-readable database name used in diagnostics.
    fn db_name() -> &'static str;

    /// Canonical database file path.
    fn db_path() -> Result<PathBuf>;

    /// Ordered embedded migration SQL files as `(id, sql)` pairs.
    fn migrations() -> &'static [(&'static str, &'static str)];

    /// Config-file lookup key under `policies.database_retry`.
    /// One of `"local_db"`, `"agent_trace_db"`, `"auth_db"`.
    fn db_config_key() -> &'static str;
}

/// Collect common filesystem health problems for a Turso database path.
pub fn collect_db_path_health(db_name: &str, db_path: &Path, problems: &mut Vec<HealthProblem>) {
    let db_name_title = sentence_case(db_name);

    let Some(parent) = db_path.parent() else {
        problems.push(HealthProblem {
            kind: HealthProblemKind::UnableToResolveStateRoot,
            category: HealthCategory::GlobalState,
            severity: HealthSeverity::Error,
            fixability: HealthFixability::ManualOnly,
            summary: format!(
                "Unable to resolve parent directory for {db_name} path '{}'.",
                db_path.display()
            ),
            remediation: String::from("Verify that the current platform exposes a writable SCE state directory before rerunning 'sce doctor'."),
            next_action: "manual_steps",
        });
        return;
    };

    if !parent.exists() {
        problems.push(HealthProblem {
            kind: HealthProblemKind::UnableToResolveStateRoot,
            category: HealthCategory::GlobalState,
            severity: HealthSeverity::Error,
            fixability: HealthFixability::AutoFixable,
            summary: format!(
                "{db_name_title} parent directory '{}' does not exist.",
                parent.display()
            ),
            remediation: format!(
                "Run 'sce doctor --fix' to create the canonical {db_name} parent directory at '{}'.",
                parent.display()
            ),
            next_action: "doctor_fix",
        });
    } else if !parent.is_dir() {
        problems.push(HealthProblem {
            kind: HealthProblemKind::UnableToResolveStateRoot,
            category: HealthCategory::GlobalState,
            severity: HealthSeverity::Error,
            fixability: HealthFixability::ManualOnly,
            summary: format!(
                "{db_name_title} parent path '{}' is not a directory.",
                parent.display()
            ),
            remediation: format!(
                "Replace '{}' with a writable directory before rerunning 'sce doctor'.",
                parent.display()
            ),
            next_action: "manual_steps",
        });
    }

    if db_path.exists() && !db_path.is_file() {
        problems.push(HealthProblem {
            kind: HealthProblemKind::UnableToResolveStateRoot,
            category: HealthCategory::GlobalState,
            severity: HealthSeverity::Error,
            fixability: HealthFixability::ManualOnly,
            summary: format!(
                "{db_name_title} path '{}' is not a file.",
                db_path.display()
            ),
            remediation: format!(
                "Replace '{}' with a writable {db_name} file path before rerunning 'sce doctor'.",
                db_path.display()
            ),
            next_action: "manual_steps",
        });
    }
}

/// Create the parent directory for a Turso database path.
pub fn bootstrap_db_parent(db_name: &str, db_path: &Path) -> Result<PathBuf> {
    let parent = db_path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("{db_name} path has no parent: {}", db_path.display()))?;

    fs::create_dir_all(parent).with_context(|| {
        format!(
            "failed to create {db_name} parent directory: {}",
            parent.display()
        )
    })?;

    Ok(parent.to_path_buf())
}

fn sentence_case(value: &str) -> String {
    let mut chars = value.chars();
    let Some(first) = chars.next() else {
        return String::new();
    };

    first.to_uppercase().collect::<String>() + chars.as_str()
}

fn ensure_db_parent_dir(db_name: &str, db_path: &Path) -> Result<()> {
    if let Some(parent) = db_path.parent() {
        std::fs::create_dir_all(parent).with_context(|| {
            format!(
                "failed to create {db_name} parent directory: {}",
                parent.display()
            )
        })?;
    }

    Ok(())
}

fn build_current_thread_runtime(db_name: &str) -> Result<tokio::runtime::Runtime> {
    tokio::runtime::Builder::new_current_thread()
        .enable_io()
        .enable_time()
        .build()
        .with_context(|| {
            format!("failed to create {db_name} tokio runtime. Try: rerun the command; if the issue persists, verify the local Tokio runtime environment.")
        })
}

fn run_embedded_migrations(
    conn: &turso::Connection,
    runtime: &tokio::runtime::Runtime,
    db_name: &str,
    migrations: &[(&str, &str)],
) -> Result<()> {
    ensure_migrations_table(conn, runtime, db_name)?;

    for (id, sql) in migrations {
        if is_migration_applied(conn, runtime, db_name, id)? {
            continue;
        }

        apply_migration(conn, runtime, db_name, id, sql)?;
    }

    Ok(())
}

fn ensure_migrations_table(
    conn: &turso::Connection,
    runtime: &tokio::runtime::Runtime,
    db_name: &str,
) -> Result<()> {
    runtime.block_on(async {
        conn.execute(MIGRATIONS_TABLE_SQL, ())
            .await
            .map_err(|e| anyhow::anyhow!("{db_name} migration metadata setup failed: {e}"))
    })?;

    Ok(())
}

fn is_migration_applied(
    conn: &turso::Connection,
    runtime: &tokio::runtime::Runtime,
    db_name: &str,
    id: &str,
) -> Result<bool> {
    runtime.block_on(async {
        let mut rows = conn.query(SELECT_MIGRATION_SQL, (id,)).await.map_err(|e| {
            anyhow::anyhow!("{db_name} migration metadata query failed for {id}: {e}")
        })?;

        rows.next().await.map(|row| row.is_some()).map_err(|e| {
            anyhow::anyhow!("{db_name} migration metadata row fetch failed for {id}: {e}")
        })
    })
}

fn apply_migration(
    conn: &turso::Connection,
    runtime: &tokio::runtime::Runtime,
    db_name: &str,
    id: &str,
    sql: &str,
) -> Result<()> {
    runtime.block_on(async {
        conn.execute(sql, ())
            .await
            .map_err(|e| anyhow::anyhow!("{db_name} migration {id} failed: {e}"))?;
        conn.execute(INSERT_MIGRATION_SQL, (id,))
            .await
            .map_err(|e| {
                anyhow::anyhow!("{db_name} migration metadata record failed for {id}: {e}")
            })?;

        Ok(())
    })
}

struct TursoConnectionCore<M: DbSpec> {
    conn: turso::Connection,
    runtime: tokio::runtime::Runtime,
    spec: PhantomData<fn() -> M>,
}

impl<M: DbSpec> TursoConnectionCore<M> {
    fn new(conn: turso::Connection, runtime: tokio::runtime::Runtime) -> Self {
        Self {
            conn,
            runtime,
            spec: PhantomData,
        }
    }

    fn run_migrations(&self) -> Result<()> {
        run_embedded_migrations(&self.conn, &self.runtime, M::db_name(), M::migrations())
    }
}

fn resolve_connection_open_retry_policy<M: DbSpec>() -> RetryPolicy {
    if let Some(config) = crate::services::config::get_database_retry_config() {
        let per_db = match M::db_config_key() {
            "local_db" => config.local_db.as_ref(),
            "agent_trace_db" => config.agent_trace_db.as_ref(),
            "auth_db" => config.auth_db.as_ref(),
            _ => None,
        };
        if let Some(per_db) = per_db {
            if let Some(policy) = per_db.connection_open {
                return policy;
            }
        }
    }
    CONNECTION_OPEN_RETRY_POLICY
}

fn resolve_query_retry_policy<M: DbSpec>() -> RetryPolicy {
    if let Some(config) = crate::services::config::get_database_retry_config() {
        let per_db = match M::db_config_key() {
            "local_db" => config.local_db.as_ref(),
            "agent_trace_db" => config.agent_trace_db.as_ref(),
            "auth_db" => config.auth_db.as_ref(),
            _ => None,
        };
        if let Some(per_db) = per_db {
            if let Some(policy) = per_db.query {
                return policy;
            }
        }
    }
    QUERY_RETRY_POLICY
}

/// Generic Turso database adapter.
///
/// Wraps a Turso connection with a tokio current-thread runtime so callers can
/// use synchronous `execute`/`query` methods while the underlying Turso API
/// remains async.
pub struct TursoDb<M: DbSpec> {
    core: TursoConnectionCore<M>,
}

/// Fully fetched SQL query result for deterministic rendering outside the
/// async Turso row iterator lifetime.
#[derive(Clone, Debug, PartialEq)]
#[allow(dead_code)]
pub struct QueryRows {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<TursoValue>>,
}

/// Generic encrypted Turso database adapter.
///
/// Mirrors the structural seams of [`TursoDb`] while reserving encrypted local
/// database initialization for services that require at-rest encryption.
pub struct EncryptedTursoDb<M: DbSpec> {
    core: TursoConnectionCore<M>,
}

impl<M: DbSpec> TursoDb<M> {
    /// Open or create the database at the spec-provided canonical path.
    ///
    /// Parent directories are created automatically. Migrations are run after
    /// the database connection is established.
    pub fn new() -> Result<Self> {
        let db = Self::open_without_migrations()?;

        db.run_migrations()
            .with_context(|| format!("failed to run {} migrations", M::db_name()))?;

        Ok(db)
    }

    /// Open or create the database at an explicit path.
    ///
    /// Parent directories are created automatically. Migrations are run after
    /// the database connection is established. The service-specific retry and
    /// migration configuration still comes from `M`.
    pub fn new_at(db_path: impl AsRef<Path>) -> Result<Self> {
        let db = Self::open_without_migrations_at(db_path)?;

        db.run_migrations()
            .with_context(|| format!("failed to run {} migrations", M::db_name()))?;

        Ok(db)
    }

    /// Open or create the database at the spec-provided canonical path without
    /// running embedded migrations.
    ///
    /// Parent directories are created automatically and the connection-open
    /// retry policy is preserved. Runtime callers that use this path are
    /// responsible for verifying schema readiness before query/write work.
    pub fn open_without_migrations() -> Result<Self> {
        let db_name = M::db_name();
        let db_path = M::db_path().with_context(|| format!("failed to resolve {db_name} path"))?;

        Self::open_without_migrations_at(db_path)
    }

    /// Open or create the database at an explicit path without running embedded
    /// migrations.
    ///
    /// Parent directories are created automatically and the connection-open
    /// retry policy is preserved. Runtime callers that use this path are
    /// responsible for verifying schema readiness before query/write work.
    pub fn open_without_migrations_at(db_path: impl AsRef<Path>) -> Result<Self> {
        let db_name = M::db_name();
        let db_path = db_path.as_ref().to_path_buf();

        ensure_db_parent_dir(db_name, &db_path)?;

        let runtime = build_current_thread_runtime(db_name)?;
        let retry_policy = resolve_connection_open_retry_policy::<M>();
        let operation_name = format!("open {db_name} database connection");

        let conn = run_with_retry_sync(
            retry_policy,
            &operation_name,
            CONNECTION_OPEN_RETRY_HINT,
            |_| {
                runtime.block_on(async {
                    let path_str = db_path.to_str().ok_or_else(|| {
                        anyhow::anyhow!("invalid UTF-8 in database path: {}", db_path.display())
                    })?;
                    let db = turso::Builder::new_local(path_str)
                        .experimental_multiprocess_wal(true)
                        .build()
                        .await
                        .map_err(|e| {
                            anyhow::anyhow!(
                                "failed to open {db_name} database at {}: {e}",
                                db_path.display()
                            )
                        })?;
                    db.connect().map_err(|e| {
                        anyhow::anyhow!("failed to connect to {db_name} database: {e}")
                    })
                })
            },
        )?;

        Ok(Self {
            core: TursoConnectionCore::new(conn, runtime),
        })
    }

    /// Execute a SQL statement that does not return rows.
    ///
    /// # Arguments
    /// * `sql` - SQL statement, which may contain `?` placeholders.
    /// * `params` - Parameter values implementing `IntoParams`.
    ///
    /// # Returns
    /// Number of rows affected.
    pub fn execute(&self, sql: &str, params: impl turso::params::IntoParams) -> Result<u64> {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("execute {} database query", M::db_name());

        run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    self.core
                        .conn
                        .execute(sql, params.clone())
                        .await
                        .map_err(|e| anyhow::anyhow!("{} execute failed: {sql}: {e}", M::db_name()))
                })
            },
        )
    }

    /// Execute a SQL query that returns rows.
    ///
    /// # Arguments
    /// * `sql` - SQL query, which may contain `?` placeholders.
    /// * `params` - Parameter values implementing `IntoParams`.
    ///
    /// # Returns
    /// A `turso::Rows` iterator over the result set.
    #[allow(dead_code)]
    pub fn query(&self, sql: &str, params: impl turso::params::IntoParams) -> Result<turso::Rows> {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("query {} database", M::db_name());

        run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    self.core
                        .conn
                        .query(sql, params.clone())
                        .await
                        .map_err(|e| anyhow::anyhow!("{} query failed: {sql}: {e}", M::db_name()))
                })
            },
        )
    }

    /// Execute a SQL query and synchronously fetch column names plus raw values.
    #[allow(dead_code)]
    pub fn query_values(
        &self,
        sql: &str,
        params: impl turso::params::IntoParams,
    ) -> Result<QueryRows> {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("query and fetch {} database values", M::db_name());

        run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    let mut rows =
                        self.core
                            .conn
                            .query(sql, params.clone())
                            .await
                            .map_err(|e| {
                                anyhow::anyhow!("{} query failed: {sql}: {e}", M::db_name())
                            })?;
                    let columns = rows.column_names();
                    let column_count = rows.column_count();
                    let mut fetched_rows = Vec::new();

                    while let Some(row) = rows.next().await.map_err(|e| {
                        anyhow::anyhow!("{} row fetch failed: {sql}: {e}", M::db_name())
                    })? {
                        let mut values = Vec::with_capacity(column_count);
                        for column_index in 0..column_count {
                            values.push(row.get_value(column_index).map_err(|e| {
                                anyhow::anyhow!("{} value fetch failed: {sql}: {e}", M::db_name())
                            })?);
                        }
                        fetched_rows.push(values);
                    }

                    Ok(QueryRows {
                        columns,
                        rows: fetched_rows,
                    })
                })
            },
        )
    }

    /// Execute a SQL query and synchronously map all returned rows.
    pub fn query_map<T, F>(
        &self,
        sql: &str,
        params: impl turso::params::IntoParams,
        mut map_row: F,
    ) -> Result<Vec<T>>
    where
        F: FnMut(&turso::Row) -> Result<T>,
    {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("query and fetch {} database rows", M::db_name());

        let rows = run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    let mut rows =
                        self.core
                            .conn
                            .query(sql, params.clone())
                            .await
                            .map_err(|e| {
                                anyhow::anyhow!("{} query failed: {sql}: {e}", M::db_name())
                            })?;
                    let mut fetched_rows = Vec::new();

                    while let Some(row) = rows.next().await.map_err(|e| {
                        anyhow::anyhow!("{} row fetch failed: {sql}: {e}", M::db_name())
                    })? {
                        fetched_rows.push(row);
                    }

                    Ok(fetched_rows)
                })
            },
        )?;

        let mut results = Vec::new();

        for row in rows {
            results.push(
                map_row(&row)
                    .with_context(|| format!("{} row mapping failed: {sql}", M::db_name()))?,
            );
        }

        Ok(results)
    }

    /// Run all embedded migrations in order.
    ///
    /// Applied migration IDs are recorded in `__sce_migrations` so later
    /// initializations apply only migrations that were not already recorded.
    /// Existing databases without migration metadata are brought forward by
    /// re-applying the current idempotent migration set and recording each ID.
    pub fn run_migrations(&self) -> Result<()> {
        self.core.run_migrations()
    }

    /// Check migration metadata for problems that would prevent safe hook
    /// runtime access.
    ///
    /// Returns a list of problems: missing migration metadata table,
    /// incomplete applied migrations, or unexpected extra migrations.
    /// An empty list means the schema is ready.
    pub fn migration_metadata_problems(&self) -> Result<Vec<String>> {
        let migration_table_exists = self.query_map(
            "SELECT name FROM sqlite_master WHERE type = 'table' AND name = '__sce_migrations' LIMIT 1",
            (),
            |row| row.get::<String>(0).map_err(Into::into),
        )?;

        if migration_table_exists.is_empty() {
            return Ok(vec![String::from("missing migration metadata table")]);
        }

        let applied_ids = self.query_map(
            "SELECT id FROM __sce_migrations ORDER BY id ASC",
            (),
            |row| row.get::<String>(0).map_err(Into::into),
        )?;
        let expected_ids = M::migrations()
            .iter()
            .map(|(id, _)| *id)
            .collect::<Vec<_>>();
        let mut problems = Vec::new();

        if applied_ids.len() != expected_ids.len() {
            problems.push(format!(
                "expected {} applied migrations, found {}",
                expected_ids.len(),
                applied_ids.len()
            ));
        }

        let missing_ids = expected_ids
            .iter()
            .copied()
            .filter(|id| !applied_ids.iter().any(|applied_id| applied_id == id))
            .collect::<Vec<_>>();
        if !missing_ids.is_empty() {
            problems.push(format!("missing migrations {}", missing_ids.join(", ")));
        }

        let unexpected_ids = applied_ids
            .iter()
            .filter(|applied_id| !expected_ids.iter().any(|id| id == &applied_id.as_str()))
            .map(String::as_str)
            .collect::<Vec<_>>();
        if !unexpected_ids.is_empty() {
            problems.push(format!(
                "unexpected migrations {}",
                unexpected_ids.join(", ")
            ));
        }

        Ok(problems)
    }

    /// Verify that the database schema needed by hook runtime readers and
    /// writers already exists.
    ///
    /// This check is intentionally non-mutating. Missing or incomplete schema
    /// is reported with the provided setup guidance instead of running
    /// migrations from a high-frequency hook path.
    pub fn ensure_schema_ready(&self, setup_guidance: &str) -> Result<()> {
        let problems = self.migration_metadata_problems()?;

        if problems.is_empty() {
            return Ok(());
        }

        anyhow::bail!(
            "{} schema is not initialized or is incomplete: {}. {setup_guidance}",
            M::db_name(),
            problems.join(", ")
        )
    }
}

impl<M: DbSpec> EncryptedTursoDb<M> {
    /// Open or create the encrypted database at the spec-provided canonical
    /// path.
    ///
    /// This constructor is the encrypted counterpart to [`TursoDb::new`] and
    /// uses a strict encrypted local-builder path.
    pub fn new() -> Result<Self> {
        let db_name = M::db_name();
        let db_path = M::db_path().with_context(|| format!("failed to resolve {db_name} path"))?;
        let encryption_key = encryption_key::get_or_create_encryption_key(&db_path, db_name)?;

        ensure_db_parent_dir(db_name, &db_path)?;

        let runtime = build_current_thread_runtime(db_name)?;
        let retry_policy = resolve_connection_open_retry_policy::<M>();
        let operation_name = format!("open encrypted {db_name} database connection");

        let conn = run_with_retry_sync(
            retry_policy,
            &operation_name,
            CONNECTION_OPEN_RETRY_HINT,
            |_| {
                runtime.block_on(async {
                    let path_str = db_path.to_str().ok_or_else(|| {
                        anyhow::anyhow!("invalid UTF-8 in database path: {}", db_path.display())
                    })?;

                    let encryption_opts = turso::EncryptionOpts {
                        hexkey: encryption_key.clone(),
                        cipher: ENCRYPTION_CIPHER_AEGIS256.to_string(),
                    };

                    let db = turso::Builder::new_local(path_str)
                        .experimental_encryption(true)
                        .with_encryption(encryption_opts)
                        .build()
                        .await
                        .map_err(|e| {
                            anyhow::anyhow!(
                                "failed to open encrypted {db_name} database at {} with cipher {ENCRYPTION_CIPHER_AEGIS256}. Try: verify the credential store encryption key is valid and that local Turso encryption support is available: {e}",
                                db_path.display()
                            )
                        })?;

                    db.connect().map_err(|e| {
                        anyhow::anyhow!("failed to connect to encrypted {db_name} database: {e}")
                    })
                })
            },
        )?;

        let db = Self {
            core: TursoConnectionCore::new(conn, runtime),
        };

        db.run_migrations()
            .with_context(|| format!("failed to run {db_name} migrations"))?;

        Ok(db)
    }

    /// Execute a SQL statement that does not return rows.
    ///
    /// # Arguments
    /// * `sql` - SQL statement, which may contain `?` placeholders.
    /// * `params` - Parameter values implementing `IntoParams`.
    ///
    /// # Returns
    /// Number of rows affected.
    pub fn execute(&self, sql: &str, params: impl turso::params::IntoParams) -> Result<u64> {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("execute encrypted {} database query", M::db_name());

        run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    self.core
                        .conn
                        .execute(sql, params.clone())
                        .await
                        .map_err(|e| anyhow::anyhow!("{} execute failed: {sql}: {e}", M::db_name()))
                })
            },
        )
    }

    /// Execute a SQL query that returns rows.
    ///
    /// # Arguments
    /// * `sql` - SQL query, which may contain `?` placeholders.
    /// * `params` - Parameter values implementing `IntoParams`.
    ///
    /// # Returns
    /// A `turso::Rows` iterator over the result set.
    #[allow(dead_code)]
    pub fn query(&self, sql: &str, params: impl turso::params::IntoParams) -> Result<turso::Rows> {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("query encrypted {} database", M::db_name());

        run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    self.core
                        .conn
                        .query(sql, params.clone())
                        .await
                        .map_err(|e| anyhow::anyhow!("{} query failed: {sql}: {e}", M::db_name()))
                })
            },
        )
    }

    /// Execute a SQL query and synchronously map all returned rows.
    pub fn query_map<T, F>(
        &self,
        sql: &str,
        params: impl turso::params::IntoParams,
        mut map_row: F,
    ) -> Result<Vec<T>>
    where
        F: FnMut(&turso::Row) -> Result<T>,
    {
        let params = turso::params::IntoParams::into_params(params).map_err(|e| {
            anyhow::anyhow!("{} parameter conversion failed: {sql}: {e}", M::db_name())
        })?;
        let operation_name = format!("query and fetch encrypted {} database rows", M::db_name());

        let rows = run_with_retry_sync(
            resolve_query_retry_policy::<M>(),
            &operation_name,
            QUERY_RETRY_HINT,
            |_| {
                self.core.runtime.block_on(async {
                    let mut rows =
                        self.core
                            .conn
                            .query(sql, params.clone())
                            .await
                            .map_err(|e| {
                                anyhow::anyhow!("{} query failed: {sql}: {e}", M::db_name())
                            })?;
                    let mut fetched_rows = Vec::new();

                    while let Some(row) = rows.next().await.map_err(|e| {
                        anyhow::anyhow!("{} row fetch failed: {sql}: {e}", M::db_name())
                    })? {
                        fetched_rows.push(row);
                    }

                    Ok(fetched_rows)
                })
            },
        )?;

        let mut results = Vec::new();

        for row in rows {
            results.push(
                map_row(&row)
                    .with_context(|| format!("{} row mapping failed: {sql}", M::db_name()))?,
            );
        }

        Ok(results)
    }

    /// Run all embedded migrations in order.
    ///
    /// Applied migration IDs are recorded in `__sce_migrations` so later
    /// initializations apply only migrations that were not already recorded.
    /// Existing databases without migration metadata are brought forward by
    /// re-applying the current idempotent migration set and recording each ID.
    pub fn run_migrations(&self) -> Result<()> {
        self.core.run_migrations()
    }
}

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

    const QUERY_RETRY_FAILURE_BUDGET_MS: u64 = 2_000;

    fn worst_case_retry_failure_budget_ms(policy: RetryPolicy) -> u64 {
        let attempt_timeouts = policy
            .timeout_ms
            .saturating_mul(u64::from(policy.max_attempts));
        let retry_backoffs = (2..=policy.max_attempts)
            .map(|attempt| retry_backoff_ms(policy, attempt))
            .fold(0_u64, u64::saturating_add);

        attempt_timeouts.saturating_add(retry_backoffs)
    }

    fn retry_backoff_ms(policy: RetryPolicy, attempt: u32) -> u64 {
        if attempt <= 1 {
            return 0;
        }

        let exponent = (attempt - 2).min(20);
        let multiplier = 1_u64 << exponent;

        policy
            .initial_backoff_ms
            .saturating_mul(multiplier)
            .min(policy.max_backoff_ms)
    }

    #[test]
    fn default_query_retry_policy_stays_within_two_second_failure_budget() {
        let budget_ms = worst_case_retry_failure_budget_ms(QUERY_RETRY_POLICY);

        assert!(
            budget_ms <= QUERY_RETRY_FAILURE_BUDGET_MS,
            "default query retry failure budget was {budget_ms}ms; expected <= {QUERY_RETRY_FAILURE_BUDGET_MS}ms"
        );
    }
}