boxdb 0.7.0

//! An extended SQLite client. Binary files are ordinary SQLite databases;
//! readable files independently store schemas and rows as a YAML stream.
//!
//! ```
//! use boxdb::{Database, Value};
//! let mut db = Database::memory();
//! db.execute("CREATE TABLE users(id INT64 PRIMARY KEY, name TEXT NOT NULL)")?;
//! db.execute_params("INSERT INTO users(name) VALUES(?)", &[Value::Text("Ada".into())])?;
//! assert_eq!(db.execute("SELECT name FROM users")?[0].rows,
//!            vec![vec![Value::Text("Ada".into())]]);
//! # Ok::<(), boxdb::Error>(())
//! ```
use rusqlite::{Connection, OpenFlags, types::Value as SqlValue};
use serde::{Deserialize, Serialize};
use std::{
    cell::{Cell, RefCell},
    collections::HashMap,
    path::Path,
    rc::Rc,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};
mod diagnostics;
mod error;
pub use diagnostics::{Check, CheckIssue, CheckMode, CheckReport, CheckStatus};
mod recovery;
pub use error::{Error, ErrorOperation, OperationOutcome};
pub use recovery::{RecoveryReport, RejectedDocument};
pub type Result<T> = std::result::Result<T, Error>;
pub mod changeset;
pub mod migrations;
pub mod types;
pub use types::{Type, Value};
mod decimal;
mod extensions;
mod history;
mod readable;
mod schema;
mod schema_dump;
pub use schema_dump::SchemaFormat;
mod storage;
#[cfg(doctest)]
#[doc = include_str!("../tests/removed_snapshot_api.md")]
mod removed_snapshot_api {}
#[derive(Clone, Copy, Debug, PartialEq, Eq, clap::ValueEnum, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Format {
    Binary,
    Readable,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct QueryResult {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<Value>>,
    pub affected: usize,
}
/// Metadata used by language bindings without executing the statement.
#[derive(Clone, Debug)]
pub struct StatementInfo {
    pub columns: Vec<String>,
    pub parameters: Vec<Option<String>>,
    pub readonly: bool,
}
/// Options used by asynchronous language bindings while opening a connection.
#[derive(Clone, Copy, Default)]
pub struct OpenOptions {
    pub read_only: bool,
    pub busy_handler: Option<fn(i32) -> bool>,
    pub progress_handler: Option<fn() -> bool>,
}
/// Thread-safe cancellation of user SQL, allowing internal cleanup to finish.
pub struct InterruptHandle {
    sqlite: rusqlite::InterruptHandle,
    cleanup: Arc<Mutex<()>>,
}
impl InterruptHandle {
    /// Interrupt active SQL. Requests during internal cleanup are ignored.
    pub fn interrupt(&self) {
        if let Ok(_guard) = self.cleanup.try_lock() {
            self.sqlite.interrupt();
        }
    }
}
pub struct Database {
    connection: Connection,
    format: Option<Format>,
    readable: Option<storage::ReadableFile>,
    types: RefCell<schema::Types>,
    result_types: RefCell<HashMap<String, Rc<[Option<Type>]>>>,
    pager_version: Cell<u32>,
    validation_dirty: Cell<bool>,
    schema_version: Cell<i64>,
    write_transaction_ready: bool,
    poisoned: Cell<bool>,
    progress_operations: Cell<i32>,
    progress_callback: Cell<Option<fn() -> bool>>,
    internal: Arc<AtomicBool>,
    interrupt_cleanup: Arc<Mutex<()>>,
    cleanup_active: Cell<bool>,
    history: Arc<history::Shared>,
    commit_triggers: RefCell<Vec<history::triggers::CommitTrigger>>,
    savepoints: RefCell<Vec<String>>,
    savepoint_transaction: Cell<bool>,
}
struct CleanupGuard<'a> {
    active: &'a Cell<bool>,
    _lock: std::sync::MutexGuard<'a, ()>,
}
impl Drop for CleanupGuard<'_> {
    fn drop(&mut self) {
        self.active.set(false);
    }
}
struct InternalGuard(Arc<AtomicBool>, bool);
impl InternalGuard {
    fn enter(flag: &Arc<AtomicBool>) -> Self {
        let old = flag.swap(true, Ordering::Relaxed);
        Self(flag.clone(), old)
    }
}
impl Drop for InternalGuard {
    fn drop(&mut self) {
        self.0.store(self.1, Ordering::Relaxed);
    }
}
fn authorizer(db: &Connection, internal: Arc<AtomicBool>) -> Result<()> {
    use rusqlite::hooks::{AuthAction as A, Authorization as Z};
    db.authorizer(Some(move |context: rusqlite::hooks::AuthContext<'_>| {
        if internal.load(Ordering::Relaxed) {
            return Z::Allow;
        }
        if let A::Function { function_name } = context.action
            && function_name.starts_with("__boxdb_instant_write_")
            && !context
                .accessor
                .is_some_and(|name| name.starts_with("__boxdb_validate_instant_"))
        {
            return Z::Deny;
        }
        let object = match context.action {
            A::Insert { table_name }
            | A::Update { table_name, .. }
            | A::Delete { table_name }
            | A::CreateTable { table_name }
            | A::DropTable { table_name }
            | A::CreateTempTable { table_name }
            | A::DropTempTable { table_name } => Some(table_name),
            A::AlterTable { table_name, .. } => Some(table_name),
            A::CreateTrigger { trigger_name, .. }
            | A::CreateTempTrigger { trigger_name, .. }
            | A::DropTrigger { trigger_name, .. }
            | A::DropTempTrigger { trigger_name, .. } => Some(trigger_name),
            _ => None,
        };
        if object.is_some_and(|n| n.to_ascii_lowercase().starts_with("__boxdb_")) {
            Z::Deny
        } else {
            Z::Allow
        }
    }))?;
    Ok(())
}
fn configure(db: &Connection, binary: bool, options: OpenOptions) -> Result<()> {
    extensions::register(db)?;
    if let Some(callback) = options.busy_handler {
        db.busy_handler(Some(callback))?;
    } else {
        db.busy_timeout(Duration::from_secs(5))?;
    }
    db.progress_handler(1000, options.progress_handler)?;
    db.pragma_update(None, "foreign_keys", true)?;
    db.pragma_update(None, "synchronous", "FULL")?;
    db.pragma_update(None, "fullfsync", true)?;
    if binary && !options.read_only {
        db.pragma_update(None, "journal_mode", "WAL")?;
        db.pragma_update(None, "locking_mode", "NORMAL")?;
    }
    db.set_prepared_statement_cache_capacity(512);
    schema::configure_codec(db, options.read_only)?;
    Ok(())
}
impl Database {
    /// Export user schema, excluding internal tables, application rows and sequences.
    pub fn dump_schema(&self, format: SchemaFormat) -> Result<Vec<u8>> {
        schema_dump::dump(&self.connection, format)
    }

    /// Export committed schema without opening a writer or changing the source.
    pub fn dump_schema_file(path: impl AsRef<Path>, format: SchemaFormat) -> Result<Vec<u8>> {
        schema_dump::dump(&storage::reader(path.as_ref())?, format)
    }

    fn new(connection: Connection, binary: bool) -> Result<Self> {
        Self::new_with_options(connection, binary, OpenOptions::default())
    }
    fn new_with_options(
        connection: Connection,
        binary: bool,
        options: OpenOptions,
    ) -> Result<Self> {
        configure(&connection, binary, options)?;
        let history = Arc::new(history::Shared::default());
        // Load the schema, validators and version markers from one snapshot.
        // A concurrent commit after this snapshot will invalidate its markers.
        let snapshot = connection.unchecked_transaction()?;
        let types = schema::load(&connection)?;
        schema::install(&connection, &types)?;
        let commit_triggers = history::triggers::load(&connection)?;
        history
            .triggered
            .store(!commit_triggers.is_empty(), Ordering::Relaxed);
        let schema_version =
            connection.query_row("PRAGMA main.schema_version", [], |r| r.get(0))?;
        let pager_version = storage::data_version(&connection)?;
        snapshot.commit()?;
        let internal = Arc::new(AtomicBool::new(false));
        authorizer(&connection, internal.clone())?;
        history::register(&connection, history.clone())?;
        history::install_hook(&connection, history.clone())?;
        Ok(Self {
            connection,
            format: None,
            readable: None,
            types: RefCell::new(types),
            result_types: RefCell::new(HashMap::new()),
            pager_version: Cell::new(pager_version),
            validation_dirty: Cell::new(false),
            schema_version: Cell::new(schema_version),
            write_transaction_ready: false,
            poisoned: Cell::new(false),
            progress_operations: Cell::new(1000),
            progress_callback: Cell::new(options.progress_handler),
            internal,
            interrupt_cleanup: Arc::new(Mutex::new(())),
            cleanup_active: Cell::new(false),
            history,
            commit_triggers: RefCell::new(commit_triggers),
            savepoints: RefCell::new(vec![]),
            savepoint_transaction: Cell::new(false),
        })
    }
    pub fn memory() -> Self {
        Self::try_memory().expect("configure BoxDB memory database")
    }
    /// Open an isolated memory database, returning initialization failures.
    pub fn try_memory() -> Result<Self> {
        Self::new(Connection::open_in_memory()?, false)
    }
    /// Open an ordinary SQLite binary database or a standalone YAML stream.
    /// Existing contents determine the format; `format` applies to new files.
    pub fn open(path: impl AsRef<Path>, format: Format) -> Result<Self> {
        Self::open_with_options(path, format, OpenOptions::default())
    }
    pub fn memory_with_options(options: OpenOptions) -> Result<Self> {
        let db = Self::new_with_options(Connection::open_in_memory()?, false, options)?;
        if options.read_only {
            db.connection.pragma_update(None, "query_only", true)?;
        }
        Ok(db)
    }
    pub fn open_with_options(
        path: impl AsRef<Path>,
        format: Format,
        options: OpenOptions,
    ) -> Result<Self> {
        Self::open_inner(path.as_ref(), format, options).map_err(|error| {
            error.state(
                ErrorOperation::Open,
                OperationOutcome::Unknown,
                false,
                false,
            )
        })
    }
    fn open_inner(path: &Path, format: Format, options: OpenOptions) -> Result<Self> {
        let detected = if path.exists() && path.metadata()?.len() > 0 {
            use std::io::Read;
            let mut header = [0u8; 16];
            let mut file = std::fs::File::open(path)?;
            let count = file.read(&mut header)?;
            if header.starts_with(b"BOXDB\0") {
                return Err(Error::new(
                    "obsolete BoxDB binary format is unsupported".into(),
                ));
            }
            if count == 16 && header == storage::SQLITE_HEADER {
                Format::Binary
            } else {
                Format::Readable
            }
        } else {
            format
        };
        let mut db = match detected {
            Format::Binary => {
                let connection = if options.read_only {
                    Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?
                } else {
                    Connection::open(path)?
                };
                Self::new_with_options(connection, true, options)?
            }
            Format::Readable => {
                if options.read_only {
                    let connection = storage::import(&std::fs::read(path)?)?;
                    let mut db = Self::new_with_options(connection, false, options)?;
                    db.connection.pragma_update(None, "query_only", true)?;
                    db.format = Some(detected);
                    return Ok(db);
                }
                let file = storage::ReadableFile::open(path)?;
                let connection = if path.exists() && path.metadata()?.len() > 0 {
                    storage::import(&std::fs::read(path)?)?
                } else {
                    Connection::open_in_memory()?
                };
                let mut db = Self::new_with_options(connection, false, options)?;
                file.watch(&db.history);
                if !path.exists() || path.metadata()?.len() == 0 {
                    file.publish(&db.connection, &db.types.borrow())?;
                }
                db.readable = Some(file);
                db
            }
        };
        db.format = Some(detected);
        Ok(db)
    }
    pub fn format(&self) -> Option<Format> {
        self.format
    }
    pub fn in_transaction(&self) -> bool {
        !self.connection.is_autocommit()
    }
    /// Reject transaction control in driver SQL; handles manage this separately.
    pub fn validate_query_sql(sql: &str) -> Result<()> {
        for statement in split(sql)? {
            if matches!(
                schema::keyword(statement).to_ascii_uppercase().as_str(),
                "BEGIN" | "COMMIT" | "END" | "ROLLBACK" | "SAVEPOINT" | "RELEASE"
            ) {
                return Err(Error::new(
                    "use transaction handles for transaction control".into(),
                ));
            }
        }
        Ok(())
    }
    /// Set SQLite's query-only guard, returning its previous value.
    pub fn query_only(&self, enabled: bool) -> Result<bool> {
        if self.poisoned.get() {
            return Err(Error::new(
                "connection is unavailable; close it and check the database before reopening"
                    .into(),
            )
            .with_code("BOXDB_UNAVAILABLE"));
        }
        let previous = self
            .connection
            .pragma_query_value(None, "query_only", |row| row.get(0))?;
        self.connection.pragma_update(None, "query_only", enabled)?;
        Ok(previous)
    }
    /// Obtain a thread-safe handle for interrupting an active SQLite operation.
    pub fn interrupt_handle(&self) -> InterruptHandle {
        InterruptHandle {
            sqlite: self.connection.get_interrupt_handle(),
            cleanup: self.interrupt_cleanup.clone(),
        }
    }
    /// Replace the default five-second busy timeout with a callback.
    /// Returning false stops waiting for a SQLite lock. A later busy_timeout PRAGMA
    /// replaces this callback again.
    pub fn busy_handler(&self, callback: Option<fn(i32) -> bool>) -> Result<()> {
        self.connection.busy_handler(callback).map_err(Into::into)
    }
    /// Install a callback that interrupts SQL when it returns true.
    pub fn progress_handler(&self, operations: i32, callback: Option<fn() -> bool>) -> Result<()> {
        self.connection
            .progress_handler(operations, callback)
            .map_err(Error::from)?;
        self.progress_operations.set(operations);
        self.progress_callback
            .set(if operations > 0 { callback } else { None });
        Ok(())
    }
    /// Convert committed data into a new file without modifying the source.
    /// Existing destinations are never overwritten. Binary output uses SQLite's
    /// backup API; readable output uses the standalone YAML storage format.
    pub fn convert(
        source: impl AsRef<Path>,
        destination: impl AsRef<Path>,
        format: Format,
    ) -> Result<()> {
        let source = storage::reader(source.as_ref())?;
        extensions::register(&source)?;
        let transaction = source.unchecked_transaction()?;
        let check: String = transaction.query_row("PRAGMA quick_check", [], |r| r.get(0))?;
        if check != "ok" {
            return Err(Error::new(format!("invalid source database: {check}")));
        }
        let types = schema::load(&transaction)?;
        schema::validate_all(&transaction, &types)?;
        if transaction
            .prepare("PRAGMA foreign_key_check")?
            .query([])?
            .next()?
            .is_some()
        {
            return Err(Error::new(
                "FOREIGN KEY constraint failed in source database".into(),
            ));
        }
        let path = destination.as_ref();
        // Reserve the destination before publication; never overwrite existing data.
        let reservation = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(path)?;
        drop(reservation);
        let result = (|| {
            match format {
                Format::Binary => {
                    let mut target = Connection::open(path)?;
                    storage::copy(&transaction, &mut target)?;
                }
                Format::Readable => {
                    let file = storage::ReadableFile::open(path)?;
                    file.publish(&transaction, &types)?;
                }
            }
            Ok(())
        })();
        if result.is_err() {
            let _ = std::fs::remove_file(path);
        }
        result
    }
    fn refresh_types(&self, force: bool) -> Result<()> {
        let version = self
            .connection
            .query_row("PRAGMA main.schema_version", [], |r| r.get(0))?;
        if force || version != self.schema_version.get() {
            let _guard = InternalGuard::enter(&self.internal);
            let types = schema::load(&self.connection)?;
            self.types.replace(types);
            self.result_types.borrow_mut().clear();
            // Recreated tables can have identical logical types while losing
            // their connection-local triggers, so schema identity matters too.
            self.validation_dirty.set(true);
            self.schema_version.set(version);
        }
        Ok(())
    }
    fn refresh(&self, force: bool) -> Result<()> {
        let pager_version = self.with_read_snapshot(|| {
            self.refresh_types(force)?;
            if self.validation_dirty.get() {
                let query_only: bool =
                    self.connection
                        .pragma_query_value(None, "query_only", |row| row.get(0))?;
                if !query_only {
                    let _guard = InternalGuard::enter(&self.internal);
                    schema::install(&self.connection, &self.types.borrow())?;
                    self.validation_dirty.set(false);
                }
            }
            let version = storage::data_version(&self.connection)?;
            if force || version != self.pager_version.get() {
                // Commit triggers live in metadata rows, so other connections'
                // declarations arrive with the pager version, not the schema.
                self.set_commit_triggers(history::triggers::load(&self.connection)?);
            }
            Ok(version)
        })?;
        self.pager_version.set(pager_version);
        Ok(())
    }
    fn with_read_snapshot<T>(&self, operation: impl FnOnce() -> Result<T>) -> Result<T> {
        if !self.connection.is_autocommit() {
            return operation();
        }
        let result = (|| {
            self.connection.execute_batch("BEGIN")?;
            let value = operation()?;
            self.connection.execute_batch("COMMIT")?;
            Ok(value)
        })();
        result.map_err(|error: Error| {
            if !self.connection.is_autocommit()
                && let Err(cleanup) = self.with_progress_paused(|| {
                    self.connection
                        .execute_batch("ROLLBACK")
                        .map_err(Into::into)
                })
            {
                self.poisoned.set(true);
                return error.cleanup(cleanup);
            }
            // TEMP schema changes may have been rolled back with the snapshot.
            self.validation_dirty.set(true);
            error
        })
    }
    fn with_progress_paused<T>(&self, operation: impl FnOnce() -> Result<T>) -> Result<T> {
        if self.cleanup_active.get() {
            return operation();
        }
        // Bindings may repeat sqlite3_interrupt until the cancelled request
        // returns. Serialize with those calls as well as pausing callbacks, so
        // an interrupt cannot turn a successful rollback into an unusable DB.
        let lock = self.interrupt_cleanup.lock().map_err(|_| {
            self.poisoned.set(true);
            Error::internal("connection interruption gate is poisoned")
        })?;
        self.cleanup_active.set(true);
        let _interrupts = CleanupGuard {
            active: &self.cleanup_active,
            _lock: lock,
        };
        self.connection.progress_handler(0, None::<fn() -> bool>)?;
        let result = operation();
        match self
            .connection
            .progress_handler(self.progress_operations.get(), self.progress_callback.get())
        {
            Ok(()) => result,
            Err(restore) => {
                self.poisoned.set(true);
                Err(match result {
                    Ok(_) => restore.into(),
                    Err(error) => error.cleanup(restore.into()),
                })
            }
        }
    }
    pub fn execute(&mut self, sql: &str) -> Result<Vec<QueryResult>> {
        self.execute_with_hook(sql, |_| Ok(()))
    }
    /// Execute a script, calling a hook before each statement. Bindings use this
    /// to restore cancellation handlers after a PRAGMA replaces the busy handler.
    pub fn execute_with_hook(
        &mut self,
        sql: &str,
        mut before_statement: impl FnMut(&Self) -> Result<()>,
    ) -> Result<Vec<QueryResult>> {
        if !sql.contains(';') {
            return if sql.trim().is_empty() {
                Ok(vec![])
            } else {
                before_statement(self)?;
                Ok(vec![self.execute_one(sql, &[], true)?])
            };
        }
        let mut results = vec![];
        for statement in split(sql)? {
            let result =
                before_statement(self).and_then(|()| self.execute_one(statement, &[], true));
            match result {
                Ok(result) => results.push(result),
                Err(error) => {
                    return Err(if results.is_empty() {
                        error
                    } else {
                        error.script_failure()
                    });
                }
            }
        }
        Ok(results)
    }
    pub fn execute_params(&mut self, sql: &str, params: &[Value]) -> Result<Vec<QueryResult>> {
        self.execute_bound(sql, params, true)
            .map(|result| vec![result])
    }
    /// Execute one statement with SQLite storage values, retaining BoxDB validation.
    /// Language drivers use this when an ORM handles logical type conversions.
    pub fn execute_native(&mut self, sql: &str, params: &[Value]) -> Result<QueryResult> {
        self.execute_bound(sql, params, false)
    }
    /// Apply a changeset, or its inverse, through `boxdb_apply`. Outside a
    /// transaction the changes commit on their own, running commit triggers.
    pub fn apply_changeset(
        &mut self,
        bytes: &[u8],
        options: changeset::ApplyOptions,
    ) -> Result<changeset::ApplyReport> {
        let options = Value::Text(serde_json::to_string(&options)?);
        let sql = "SELECT boxdb_apply(?1, ?2)";
        let parameters = [Value::Changeset(bytes.to_vec()), options];
        let result = if self.in_transaction() {
            self.execute_params(sql, &parameters)
        } else {
            self.execute("BEGIN IMMEDIATE")?;
            match self.execute_params(sql, &parameters) {
                Ok(result) => self.execute("COMMIT").map(|_| result),
                Err(error) => {
                    if self.in_transaction()
                        && let Err(rollback) = self.execute("ROLLBACK")
                    {
                        return Err(error.cleanup(rollback));
                    }
                    Err(error)
                }
            }
        }?;
        match result
            .first()
            .and_then(|r| r.rows.first())
            .and_then(|r| r.first())
        {
            Some(Value::Text(json)) => Ok(serde_json::from_str(json)?),
            _ => Err(Error::internal("boxdb_apply returned no report")),
        }
    }
    /// The rowid of the last successful INSERT on this connection.
    pub fn last_insert_rowid(&self) -> i64 {
        self.connection.last_insert_rowid()
    }
    /// Prepare a statement for validation and metadata, without running it.
    /// Execution uses the same connection's SQLite prepared statement cache.
    pub fn statement_info(&mut self, sql: &str) -> Result<StatementInfo> {
        let result = self.statement_info_inner(sql);
        result.map_err(|error| {
            self.operation_error(error, ErrorOperation::Prepare, OperationOutcome::NotApplied)
        })
    }
    fn statement_info_inner(&mut self, sql: &str) -> Result<StatementInfo> {
        if self.poisoned.get() {
            return Err(Error::new(
                "connection is unavailable; close it and check the database before reopening"
                    .into(),
            )
            .with_code("BOXDB_UNAVAILABLE"));
        }
        let statements = split(sql)?;
        let [sql] = statements.as_slice() else {
            return Err(Error::new("preparation requires one statement".into()));
        };
        self.refresh(false)?;
        if history::triggers::parse(sql)?.is_some() {
            return Ok(StatementInfo {
                columns: vec![],
                parameters: vec![],
                readonly: false,
            });
        }
        let declaration = schema::declaration(sql)?;
        let alteration = if schema::keyword(sql).eq_ignore_ascii_case("ALTER") {
            schema::alteration(sql)?
        } else {
            None
        };
        let translated = declaration
            .as_ref()
            .map(|d| d.sql.as_str())
            .or_else(|| alteration.as_ref().map(|a| a.sql.as_str()))
            .unwrap_or(sql);
        let statement = self.connection.prepare_cached(translated)?;
        Ok(StatementInfo {
            columns: statement
                .column_names()
                .into_iter()
                .map(String::from)
                .collect(),
            parameters: (1..=statement.parameter_count())
                .map(|i| statement.parameter_name(i).map(String::from))
                .collect(),
            readonly: statement.readonly(),
        })
    }
    fn execute_bound(&mut self, sql: &str, params: &[Value], typed: bool) -> Result<QueryResult> {
        let statements = split(sql)?;
        let [sql] = statements.as_slice() else {
            return Err(Error::new(
                "parameterized execution requires one statement".into(),
            ));
        };
        let values = params
            .iter()
            .map(schema::encode)
            .collect::<Result<Vec<_>>>()?;
        self.execute_one(sql, &values, typed)
    }
    fn execute_one(&mut self, sql: &str, params: &[SqlValue], typed: bool) -> Result<QueryResult> {
        let result = self.execute_one_inner(sql, params, typed);
        result.map_err(|error| {
            let operation = match schema::keyword(sql).to_ascii_uppercase().as_str() {
                "SELECT" | "VALUES" | "EXPLAIN" => ErrorOperation::Read,
                "COMMIT" | "END" => ErrorOperation::Commit,
                "ROLLBACK" => ErrorOperation::Rollback,
                _ => ErrorOperation::Write,
            };
            let outcome = if operation == ErrorOperation::Read {
                OperationOutcome::NotApplied
            } else {
                OperationOutcome::Unknown
            };
            self.operation_error(error, operation, outcome)
        })
    }
    fn operation_error(
        &mut self,
        error: Error,
        operation: ErrorOperation,
        outcome: OperationOutcome,
    ) -> Error {
        if error.fatal() {
            self.poisoned.set(true);
        }
        error.state(
            operation,
            outcome,
            self.in_transaction(),
            !self.poisoned.get(),
        )
    }
    fn cleanup_failure(&mut self, mut error: Error, sql: &str, refresh: bool) -> Error {
        let cleanup = self.with_progress_paused(|| {
            if !self.connection.is_autocommit() {
                self.connection.execute_batch(sql)?;
            }
            if refresh {
                self.refresh(true)?;
            }
            Ok(())
        });
        if let Err(cleanup) = cleanup {
            self.poisoned.set(true);
            error = error.cleanup(cleanup);
        }
        error
    }
    fn execute_one_inner(
        &mut self,
        sql: &str,
        params: &[SqlValue],
        typed: bool,
    ) -> Result<QueryResult> {
        if self.poisoned.get() {
            return Err(Error::new(
                "connection is unavailable; close it and check the database before reopening"
                    .into(),
            )
            .with_code("BOXDB_UNAVAILABLE"));
        }
        if self.connection.is_autocommit() {
            // Recording, context and trigger state belong to one transaction.
            self.history.reset();
        }
        let keyword = schema::keyword(sql);
        if keyword.eq_ignore_ascii_case("SELECT") || keyword.eq_ignore_ascii_case("VALUES") {
            return self.run(sql, params, typed);
        }
        let keyword = keyword.to_ascii_uppercase();
        let standalone_write = self.format == Some(Format::Binary)
            && self.connection.is_autocommit()
            && match keyword.as_str() {
                "SELECT" | "VALUES" | "EXPLAIN" | "BEGIN" | "SAVEPOINT" | "COMMIT" | "END"
                | "RELEASE" | "ROLLBACK" | "PRAGMA" | "VACUUM" | "ATTACH" | "DETACH" => false,
                // Preparation here is advisory. Reprepare after acquiring the
                // writer and refreshing validators against its pinned schema.
                "WITH" => !self
                    .connection
                    .prepare_cached(sql)
                    .is_ok_and(|statement| statement.readonly()),
                _ => true,
            };
        if !standalone_write {
            return self.execute_statement(sql, params, typed);
        }
        // Own the writer before reading metadata. A competing DDL commit can
        // otherwise land between refresh() and SQLite's implicit write lock.
        if let Err(error) = self.connection.execute_batch("BEGIN IMMEDIATE") {
            let error = error.into();
            return Err(if self.connection.is_autocommit() {
                error
            } else {
                self.cleanup_failure(error, "ROLLBACK", true)
            });
        }
        let result = self
            .execute_statement(sql, params, typed)
            .and_then(|result| {
                // The writer is still held, so these types and the pager marker
                // describe the same schema. A later commit must invalidate it.
                self.pager_version
                    .set(storage::data_version(&self.connection)?);
                Ok(result)
            });
        self.write_transaction_ready = false;
        let result = match result {
            Err(error)
                if error.result_code() == Some(rusqlite::ffi::SQLITE_INTERRUPT)
                    || error.fatal() =>
            {
                return Err(self.cleanup_failure(error, "ROLLBACK", true));
            }
            result => result,
        };
        if !self.connection.is_autocommit()
            && let Err(commit) = result
                .as_ref()
                .map_or(Ok(()), |_| self.before_commit())
                .and_then(|()| self.connection.execute_batch("COMMIT").map_err(Error::from))
        {
            let error = match result {
                Ok(_) => commit,
                Err(error) => error.cleanup(commit),
            };
            return Err(self.cleanup_failure(error, "ROLLBACK", true));
        }
        // Complete the implicit transaction even after a statement error:
        // SQLite's OR FAIL may have preserved earlier row changes. OR ROLLBACK
        // has already ended the transaction and needs no further cleanup.
        result
    }
    fn execute_statement(
        &mut self,
        sql: &str,
        params: &[SqlValue],
        typed: bool,
    ) -> Result<QueryResult> {
        let keyword = schema::keyword(sql);
        if keyword.eq_ignore_ascii_case("SELECT") || keyword.eq_ignore_ascii_case("VALUES") {
            return self.run(sql, params, typed);
        }
        let keyword = keyword.to_ascii_uppercase();
        let declaration = schema::declaration(sql)?;
        if let Some(declaration) = declaration {
            if declaration
                .table
                .to_ascii_lowercase()
                .starts_with("__boxdb_")
            {
                return Err(Error::new("__boxdb_ names are reserved".into()));
            }
            let _guard = InternalGuard::enter(&self.internal);
            if !params.is_empty() {
                return Err(Error::new("DDL parameters are unsupported".into()));
            }
            if self.connection.query_row(
                &format!(
                    "SELECT EXISTS(SELECT 1 FROM {}.sqlite_schema WHERE name=?1 COLLATE NOCASE)",
                    schema::quote(&declaration.schema)
                ),
                [&declaration.table],
                |r| r.get::<_, bool>(0),
            )? {
                // Preserve native IF NOT EXISTS behavior, without replacing metadata.
                self.connection.execute_batch(&declaration.sql)?;
                return Ok(empty());
            }
            self.connection
                .execute_batch("SAVEPOINT __boxdb_declaration")?;
            let result = (|| {
                declaration.execute(&self.connection)?;
                for column in &declaration.readonly {
                    schema::readonly::declare(&self.connection, &declaration.table, column)?;
                }
                schema::save(&self.connection, &declaration)?;
                self.refresh(true)?;
                self.connection
                    .execute_batch("RELEASE __boxdb_declaration")?;
                Ok(empty())
            })();
            let result = result.map_err(|error| {
                self.cleanup_failure(
                    error,
                    "ROLLBACK TO __boxdb_declaration; RELEASE __boxdb_declaration",
                    true,
                )
            });
            if result.is_ok() {
                self.publish_readable()?;
            }
            return result;
        }
        if let Some(trigger) = history::triggers::parse(sql)? {
            if !params.is_empty() {
                return Err(Error::new("DDL parameters are unsupported".into()));
            }
            return self.create_commit_trigger(trigger);
        }
        if keyword == "DROP"
            && let Some((name, if_exists)) = history::triggers::drop_statement(sql)?
            && let Some(existing) = history::triggers::find(&self.connection, &name)?
        {
            let _ = if_exists;
            return self.drop_commit_trigger(&existing);
        }
        let control = matches!(
            keyword.as_str(),
            "BEGIN" | "SAVEPOINT" | "COMMIT" | "END" | "RELEASE" | "ROLLBACK"
        );
        let ddl = matches!(keyword.as_str(), "CREATE" | "ALTER" | "DROP");
        let alteration = if keyword == "ALTER" {
            schema::alteration(sql)?
        } else {
            None
        };
        let sql = alteration.as_ref().map_or(sql, |a| a.sql.as_str());
        // Refresh validators before preparing a write: another BoxDB client may
        // have renamed/dropped a typed column referenced by our TEMP triggers.
        if !control
            && !matches!(keyword.as_str(), "SELECT" | "VALUES" | "EXPLAIN")
            && (!self.write_transaction_ready || self.connection.is_autocommit())
        {
            self.refresh(false)?;
        }
        let readonly = if ddl {
            false
        } else {
            self.connection.prepare_cached(sql)?.readonly()
        };
        let triggered = !self.commit_triggers.borrow().is_empty();
        let guarded = ddl
            || ((self.readable.is_some() || triggered)
                && self.connection.is_autocommit()
                && !readonly
                && !control
                && !matches!(keyword.as_str(), "PRAGMA" | "VACUUM" | "ATTACH" | "DETACH"));
        let outer_autocommit = self.connection.is_autocommit();
        if guarded {
            self.connection.execute_batch("SAVEPOINT __boxdb_write")?;
        }
        if triggered && !readonly && !control {
            history::ensure_recording(&self.history, &self.connection)?;
        }
        let savepoint = if control {
            history::triggers::control(sql)?
        } else {
            None
        };
        let committing = matches!(keyword.as_str(), "COMMIT" | "END")
            || matches!(&savepoint, Some(history::triggers::Control::Release(name))
                if self.savepoint_transaction.get()
                    && self.savepoints.borrow().first().is_some_and(|first| first.eq_ignore_ascii_case(name)));
        if committing && let Err(error) = self.before_commit() {
            return Err(if guarded {
                self.cleanup_failure(
                    error,
                    "ROLLBACK TO __boxdb_write; RELEASE __boxdb_write",
                    ddl,
                )
            } else {
                error
            });
        }
        if ddl {
            let _guard = InternalGuard::enter(&self.internal);
            schema::uninstall(&self.connection)?;
            if keyword == "DROP" {
                schema::readonly::before_drop(&self.connection, sql)?;
            }
        }
        // VACUUM rebuilds the complete database, including reserved metadata.
        let _vacuum_guard = (keyword == "VACUUM").then(|| InternalGuard::enter(&self.internal));
        let result = self
            .run(sql, params, typed)
            .map(|result| if ddl { empty() } else { result });
        if let Err(error) = result {
            return Err(if guarded {
                self.cleanup_failure(
                    error,
                    "ROLLBACK TO __boxdb_write; RELEASE __boxdb_write",
                    ddl,
                )
            } else {
                error
            });
        }
        let completion = (|| -> Result<()> {
            if ddl {
                let _guard = InternalGuard::enter(&self.internal);
                self.connection.flush_prepared_statement_cache();
                if matches!(keyword.as_str(), "CREATE" | "ALTER") {
                    schema::validate_table_column_names(&self.connection)?;
                }
                if self.readable.is_some()
                    && self.connection.query_row(
                        "SELECT EXISTS(SELECT 1 FROM pragma_table_list WHERE schema='main' AND type='virtual')",
                        [],
                        |row| row.get::<_, bool>(0),
                    )?
                {
                    return Err(Error::new("virtual tables require binary storage; readable storage does not support them".into()));
                }
                schema::readonly::normalize(&self.connection)?;
                if let Some(change) = &alteration {
                    for column in &change.readonly {
                        schema::readonly::declare(&self.connection, &change.table, column)?;
                    }
                    schema::save_alteration(&self.connection, change)?;
                    if !change.columns.is_empty() {
                        let types = schema::load(&self.connection)?;
                        schema::validate_all(&self.connection, &types)?;
                    }
                }
                schema::cleanup(&self.connection)?;
                schema::configure_codec(&self.connection, false)?;
                self.refresh(true)?;
                let mut recorder = history::lock(&self.history.recorder);
                if recorder.active {
                    recorder.load_tables(&self.connection)?;
                }
                drop(recorder);
                if alteration.as_ref().is_some_and(|a| !a.columns.is_empty()) {
                    schema::normalize_arrays(&self.connection, &self.types.borrow())?;
                }
            }
            if guarded {
                if outer_autocommit && !ddl {
                    self.before_commit()?;
                }
                // An internal savepoint nested inside a user transaction must
                // not publish that transaction's pending readable data.
                if outer_autocommit && let Some(file) = &self.readable {
                    storage::check_commit_constraints(&self.connection)?;
                    if let Err(e) = file.publish(&self.connection, &self.types.borrow()) {
                        self.poisoned.set(true);
                        return Err(e);
                    }
                }
                self.connection.execute_batch("RELEASE __boxdb_write")?;
            } else if !readonly || matches!(keyword.as_str(), "COMMIT" | "END" | "RELEASE") {
                self.publish_readable()?;
            }
            Ok(())
        })();
        if let Err(e) = completion {
            return Err(if guarded {
                self.cleanup_failure(e, "ROLLBACK TO __boxdb_write; RELEASE __boxdb_write", true)
            } else {
                e
            });
        }
        if control {
            self.track_savepoint(&keyword, savepoint, outer_autocommit);
            self.write_transaction_ready = false;
            self.refresh(keyword == "ROLLBACK")?;
        } else if !readonly {
            self.write_transaction_ready = true;
        }
        result
    }
    fn track_savepoint(
        &self,
        keyword: &str,
        control: Option<history::triggers::Control>,
        outer_autocommit: bool,
    ) {
        use history::triggers::Control;
        let mut savepoints = self.savepoints.borrow_mut();
        match control {
            Some(Control::Savepoint(name)) => {
                if outer_autocommit {
                    savepoints.clear();
                    self.savepoint_transaction.set(true);
                }
                savepoints.push(name);
            }
            Some(Control::Release(name)) => {
                if let Some(index) = savepoints
                    .iter()
                    .rposition(|s| s.eq_ignore_ascii_case(&name))
                {
                    savepoints.truncate(index);
                }
            }
            Some(Control::RollbackTo(name)) => {
                if let Some(index) = savepoints
                    .iter()
                    .rposition(|s| s.eq_ignore_ascii_case(&name))
                {
                    savepoints.truncate(index + 1);
                }
            }
            None => {
                if matches!(keyword, "BEGIN" | "COMMIT" | "END" | "ROLLBACK") {
                    savepoints.clear();
                    self.savepoint_transaction.set(false);
                }
            }
        }
        if self.connection.is_autocommit() {
            savepoints.clear();
            self.savepoint_transaction.set(false);
        }
    }
    fn set_commit_triggers(&self, triggers: Vec<history::triggers::CommitTrigger>) {
        self.history
            .triggered
            .store(!triggers.is_empty(), Ordering::Relaxed);
        self.commit_triggers.replace(triggers);
    }
    /// Run commit triggers inside the transaction that is about to commit.
    fn before_commit(&self) -> Result<()> {
        let triggers = self.commit_triggers.borrow().clone();
        if triggers.is_empty() || self.connection.is_autocommit() {
            return Ok(());
        }
        let mut recorder = history::lock(&self.history.recorder);
        if !recorder.active || recorder.skip_triggers || recorder.paused {
            return Ok(());
        }
        let changes = recorder.changeset(&self.connection, true)?;
        recorder.paused = true;
        drop(recorder);
        let result = (|| {
            for trigger in &triggers {
                let subset: Vec<changeset::RawChange> = changes
                    .iter()
                    .filter(|change| trigger.covers(&change.table))
                    .cloned()
                    .collect();
                if subset.is_empty() {
                    continue;
                }
                *history::lock(&self.history.stash) = Some(changeset::encode_raw(&subset, false));
                for statement in &trigger.body {
                    self.connection.execute_batch(statement).map_err(|error| {
                        Error::from(error)
                            .with_context(format!("commit trigger {} failed", trigger.name))
                    })?;
                }
            }
            Ok(())
        })();
        *history::lock(&self.history.stash) = None;
        history::lock(&self.history.recorder).paused = false;
        result
    }
    fn create_commit_trigger(
        &mut self,
        declaration: history::triggers::Declaration,
    ) -> Result<QueryResult> {
        let trigger = declaration.trigger;
        if trigger.name.to_ascii_lowercase().starts_with("__boxdb_") {
            return Err(Error::new("__boxdb_ names are reserved".into()));
        }
        self.refresh(false)?;
        let _guard = InternalGuard::enter(&self.internal);
        self.connection
            .execute_batch("SAVEPOINT __boxdb_declaration")?;
        let result = (|| {
            if history::triggers::find(&self.connection, &trigger.name)?.is_some() {
                if declaration.if_not_exists {
                    return Ok(());
                }
                return Err(Error::new(format!(
                    "commit trigger {} already exists",
                    trigger.name
                )));
            }
            if self.connection.query_row(
                "SELECT EXISTS(SELECT 1 FROM main.sqlite_schema WHERE type='trigger' AND name=?1 COLLATE NOCASE)",
                [&trigger.name],
                |row| row.get::<_, bool>(0),
            )? {
                return Err(Error::new(format!(
                    "trigger {} already exists",
                    trigger.name
                )));
            }
            for table in trigger.tables.iter().flatten() {
                if !self.connection.query_row(
                    "SELECT EXISTS(SELECT 1 FROM pragma_table_list WHERE schema='main' AND type='table' AND name=?1 COLLATE NOCASE)",
                    [table],
                    |row| row.get::<_, bool>(0),
                )? {
                    return Err(Error::new(format!("no such main table: {table}")));
                }
            }
            for statement in &trigger.body {
                self.connection
                    .prepare(statement)
                    .map_err(|error| Error::from(error).with_context("commit trigger body"))?;
            }
            history::triggers::save(&self.connection, &trigger)?;
            Ok(())
        })();
        let result = result.and_then(|()| {
            self.set_commit_triggers(history::triggers::load(&self.connection)?);
            self.connection
                .execute_batch("RELEASE __boxdb_declaration")?;
            Ok(empty())
        });
        let result = result.map_err(|error| {
            self.cleanup_failure(
                error,
                "ROLLBACK TO __boxdb_declaration; RELEASE __boxdb_declaration",
                true,
            )
        });
        if result.is_ok() {
            self.publish_readable()?;
        }
        result
    }
    fn drop_commit_trigger(&mut self, name: &str) -> Result<QueryResult> {
        let _guard = InternalGuard::enter(&self.internal);
        self.connection
            .execute_batch("SAVEPOINT __boxdb_declaration")?;
        let result = (|| {
            history::triggers::remove(&self.connection, name)?;
            self.set_commit_triggers(history::triggers::load(&self.connection)?);
            self.connection
                .execute_batch("RELEASE __boxdb_declaration")?;
            Ok(empty())
        })();
        let result = result.map_err(|error| {
            self.cleanup_failure(
                error,
                "ROLLBACK TO __boxdb_declaration; RELEASE __boxdb_declaration",
                true,
            )
        });
        if result.is_ok() {
            self.publish_readable()?;
        }
        result
    }
    fn publish_readable(&mut self) -> Result<()> {
        if self.connection.is_autocommit()
            && let Some(file) = &self.readable
            && let Err(e) = file.publish(&self.connection, &self.types.borrow())
        {
            self.poisoned.set(true);
            return Err(e);
        }
        Ok(())
    }
    fn refresh_read_metadata(&self, row_active: bool) -> Result<()> {
        let version = storage::data_version(&self.connection)?;
        if version != self.pager_version.get() {
            if row_active
                && self.connection.transaction_state(Some("main"))?
                    == rusqlite::TransactionState::None
            {
                // Constant/TEMP/attached results have no main-column origins.
                // Defer main metadata until a statement actually reads it.
                return Ok(());
            }
            let load = || {
                // Ordinary external DML changes the pager version, not the schema.
                // Keep prepared queries and validators warm in that common case.
                self.refresh_types(false)?;
                storage::data_version(&self.connection)
            };
            let version = if row_active {
                // SQLite already pins the main snapshot for this cursor.
                // BEGIN here would leave an explicit transaction whose rollback
                // can be interrupted until the outer cursor is dropped.
                load()?
            } else {
                self.with_read_snapshot(load)?
            };
            self.pager_version.set(version);
        }
        Ok(())
    }
    fn run(&self, sql: &str, params: &[SqlValue], typed: bool) -> Result<QueryResult> {
        let params: Vec<_> = params
            .iter()
            .cloned()
            .map(decimal::WireValue::new)
            .collect();
        let mut statement = self.connection.prepare_cached(sql)?;
        let before = self.connection.total_changes();
        let readonly = statement.readonly();
        let mut rows = vec![];
        if statement.column_count() == 0 {
            statement.execute(rusqlite::params_from_iter(&params))?;
        } else {
            let mut hints: Option<Rc<[Option<Type>]>> = None;
            let mut cursor = statement.query(rusqlite::params_from_iter(&params))?;
            while let Some(row) = cursor.next()? {
                // SQLite may reprepare a cached statement on its first step.
                // Read metadata after that step, including a changed SELECT *.
                if rows.is_empty() {
                    self.refresh_read_metadata(true)?;
                }
                let count = row.as_ref().column_count();
                if hints.is_none() && !self.types.borrow().is_empty() {
                    hints = self.result_types.borrow().get(sql).cloned();
                    if hints.is_none() {
                        let types = self.types.borrow();
                        let columns: Rc<[Option<Type>]> = row
                            .as_ref()
                            .columns_with_metadata()
                            .iter()
                            .map(|c| {
                                if c.database_name() != Some("main") {
                                    return None;
                                }
                                c.table_name()
                                    .zip(c.origin_name())
                                    .and_then(|(t, c)| types.get(&(t.into(), c.into())).cloned())
                            })
                            .collect();
                        let mut cache = self.result_types.borrow_mut();
                        if cache.len() >= 512 {
                            cache.clear();
                        }
                        cache.insert(sql.to_owned(), columns.clone());
                        hints = Some(columns);
                    }
                }
                let mut values = Vec::with_capacity(count);
                for i in 0..count {
                    let value = row.get::<_, SqlValue>(i)?;
                    values.push(
                        if let Some(Some(ty)) = hints.as_ref().and_then(|h| h.get(i)) {
                            if typed {
                                schema::decode(ty, value)
                                    .map_err(|error| error.with_code("BOXDB_INVALID_VALUE"))?
                            } else {
                                let decoded = schema::decode(ty, value.clone())
                                    .map_err(|error| error.with_code("BOXDB_INVALID_VALUE"))?;
                                // Storage-class results still distinguish
                                // changesets from ordinary BLOB bytes.
                                if matches!(ty, Type::Changeset) {
                                    decoded
                                } else {
                                    schema::native(value)
                                }
                            }
                        } else {
                            schema::native(value)
                        },
                    );
                }
                rows.push(values);
            }
        }
        if rows.is_empty() && readonly {
            self.refresh_read_metadata(false)?;
        }
        let columns = statement
            .column_names()
            .into_iter()
            .map(String::from)
            .collect();
        let affected = if !readonly && self.connection.total_changes() != before {
            self.connection.changes() as usize
        } else {
            0
        };
        Ok(QueryResult {
            columns,
            rows,
            affected,
        })
    }
}
fn empty() -> QueryResult {
    QueryResult {
        columns: vec![],
        rows: vec![],
        affected: 0,
    }
}
/// SQLite's completeness checker handles quoted semicolons and trigger bodies.
fn split(sql: &str) -> Result<Vec<&str>> {
    use std::ffi::CString;
    let mut statements = vec![];
    let mut start = 0;
    for (i, byte) in sql.bytes().enumerate() {
        if byte == b';' {
            let prefix = &sql[start..=i];
            let text = CString::new(prefix)
                .map_err(|error| Error::new("SQL contains NUL".into()).with_source(error))?;
            // SAFETY: CString provides a valid NUL-terminated string for this call.
            if unsafe { rusqlite::ffi::sqlite3_complete(text.as_ptr()) } != 0 {
                if !schema::keyword(prefix).is_empty() {
                    statements.push(prefix);
                }
                start = i + 1;
            }
        }
    }
    if !schema::keyword(&sql[start..]).is_empty() {
        statements.push(&sql[start..]);
    }
    Ok(statements)
}