diesel 2.4.0

use super::SqliteConnection;
use crate::query_builder::{AstPass, Query, QueryFragment, QueryId};
use crate::query_dsl::RunQueryDslSupport;
use crate::result::QueryResult;
use crate::sqlite::Sqlite;
use alloc::ffi::CString;

/// The `auto_vacuum` mode of a database, controlling whether and when SQLite
/// reclaims freed pages back to the file.
///
/// The mode is stored in the database file, not the connection. [`Full`] and
/// [`Incremental`] can be switched between at any time, but changing from or to
/// [`None`] only takes effect on a database with no tables yet, or after a
/// subsequent `VACUUM` rewrites the file.
///
/// [`None`]: AutoVacuumMode::None
/// [`Full`]: AutoVacuumMode::Full
/// [`Incremental`]: AutoVacuumMode::Incremental
#[derive(Debug, Clone, Copy, PartialEq, Eq, crate::types::Enum)]
#[diesel(sql_type = crate::sql_types::Integer)]
#[non_exhaustive]
#[repr(i32)]
pub enum AutoVacuumMode {
    /// Freed pages stay on the freelist and the file never shrinks (default).
    None = 0,
    /// Freed pages are reclaimed and the file truncated at every commit.
    Full = 1,
    /// Freelist bookkeeping is kept, pages are reclaimed only when
    /// `incremental_vacuum` runs.
    Incremental = 2,
}

/// The mode of a [`wal_checkpoint`](SqliteConnection::wal_checkpoint) run,
/// matching the modes of
/// [`sqlite3_wal_checkpoint_v2`](https://www.sqlite.org/c3ref/wal_checkpoint_v2.html).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum WalCheckpointMode {
    /// Checkpoint what is possible without waiting on readers or writers.
    Passive,
    /// Wait until there is no writer and every reader reads from the most
    /// recent snapshot, then checkpoint every frame.
    Full,
    /// Like [`Full`](Self::Full), then wait until no reader uses the WAL, so
    /// the next writer restarts the log.
    Restart,
    /// Like [`Restart`](Self::Restart), then truncate the WAL file to zero
    /// bytes.
    Truncate,
    /// Report the WAL state without checkpointing anything.
    ///
    /// Requires SQLite 3.51.0 or later. Older versions do not know this
    /// mode and silently run a [`Passive`](Self::Passive) checkpoint
    /// instead.
    Noop,
}

/// The result of a [`wal_checkpoint`](SqliteConnection::wal_checkpoint) run.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct WalCheckpointOutcome {
    /// Whether a busy reader or writer stopped the checkpoint early. Only
    /// the blocking modes set it: [`Passive`](WalCheckpointMode::Passive)
    /// reports `false` even when it left frames behind.
    pub busy: bool,
    /// Frames in the WAL after the checkpoint, `None` when the database is
    /// not in WAL mode.
    pub log_frames: Option<i64>,
    /// Frames of the WAL moved into the database file, `None` when the
    /// database is not in WAL mode. Counted within the current log, so a
    /// [`Truncate`](WalCheckpointMode::Truncate) run reports `Some(0)`
    /// because the log was emptied, not because nothing was moved.
    pub checkpointed_frames: Option<i64>,
}

impl SqliteConnection {
    /// Rebuild a database, repacking it into the smallest space it can occupy.
    ///
    /// `schema` selects an attached database by name, `None` targets `main`.
    ///
    /// This cannot run inside a transaction, needs free space of up to twice the size
    /// of the database while it runs, and renumbers the implicit `rowid` of any table
    /// declared without an `INTEGER PRIMARY KEY`.
    ///
    /// Naming a schema requires SQLite 3.24.0 or later, otherwise returns an error.
    ///
    /// ```rust
    /// # include!("../../doctest_setup.rs");
    /// # fn main() {
    /// #     run_test().unwrap();
    /// # }
    /// # fn run_test() -> QueryResult<()> {
    /// let conn = &mut SqliteConnection::establish(":memory:").unwrap();
    /// conn.vacuum(None)?;
    /// #     Ok(())
    /// # }
    /// ```
    pub fn vacuum(&mut self, schema: Option<&str>) -> QueryResult<()> {
        use crate::query_dsl::RunQueryDsl;

        Vacuum { schema, into: None }.execute(self).map(|_| ())
    }

    /// Write a vacuumed copy of a database to `path`, leaving the original untouched.
    ///
    /// This is SQLite's online backup: the copy is consistent, defragmented, and taken
    /// without blocking readers. `schema` selects an attached database by name, `None`
    /// copies `main`. The path is a bind parameter, so it needs no quoting.
    ///
    /// `path` may name a file that does not exist or one that is empty, but writing
    /// over an existing database fails rather than replacing it.
    ///
    /// Requires SQLite 3.27.0 or later, otherwise returns an error.
    ///
    /// ```rust
    /// # include!("../../doctest_setup.rs");
    /// # // This test creates a temporary directory, which is not available
    /// # // when miri runs with isolation enabled.
    /// # #[cfg(miri)]
    /// # fn main() {}
    /// # #[cfg(not(miri))]
    /// # fn main() {
    /// #     run_test().unwrap();
    /// # }
    /// # fn run_test() -> QueryResult<()> {
    /// # let dir = tempfile::tempdir().unwrap();
    /// # let backup = dir.path().join("backup.db");
    /// let conn = &mut SqliteConnection::establish(":memory:").unwrap();
    /// conn.vacuum_into(None, backup.to_str().unwrap())?;
    /// # assert!(backup.exists());
    /// #     Ok(())
    /// # }
    /// ```
    pub fn vacuum_into(&mut self, schema: Option<&str>, path: &str) -> QueryResult<()> {
        use crate::query_dsl::RunQueryDsl;

        CString::new(path)?;

        Vacuum {
            schema,
            into: Some(path),
        }
        .execute(self)
        .map(|_| ())
    }

    /// Checkpoint the [write-ahead log](https://www.sqlite.org/wal.html),
    /// moving committed frames from the WAL file into the database file.
    ///
    /// `schema` selects one attached database by name. Unlike the other
    /// maintenance helpers, `None` does not mean `main`: SQLite defines the
    /// unqualified pragma to checkpoint every attached database. With
    /// `None` and several attached databases the C API leaves the frame
    /// counts undefined.
    ///
    /// A checkpoint stopped early by a reader or writer on another
    /// connection is not an error: it sets
    /// [`busy`](WalCheckpointOutcome::busy). On a database that is not in
    /// WAL mode the call succeeds with both frame counts `None`, so it is
    /// safe to issue unconditionally. Inside a transaction on its own
    /// connection it fails with `SQLITE_LOCKED`.
    ///
    /// The mode argument requires SQLite 3.7.6 or later,
    /// [`Truncate`](WalCheckpointMode::Truncate) requires 3.8.8 or later,
    /// and [`Noop`](WalCheckpointMode::Noop) requires 3.51.0 or later.
    /// Older versions do not report an error and treat an unrecognized
    /// mode as [`Passive`](WalCheckpointMode::Passive).
    ///
    /// ```rust
    /// # include!("../../doctest_setup.rs");
    /// #
    /// # // This test creates a temporary directory, which is not available
    /// # // when miri runs with isolation enabled.
    /// # #[cfg(miri)]
    /// # fn main() {}
    /// # #[cfg(not(miri))]
    /// # fn main() {
    /// #     run_test().unwrap();
    /// # }
    /// #
    /// # fn run_test() -> QueryResult<()> {
    /// use diesel::connection::SimpleConnection;
    /// use diesel::sqlite::WalCheckpointMode;
    /// # let dir = tempfile::tempdir().unwrap();
    /// # let path = dir.path().join("app.db");
    /// let conn = &mut SqliteConnection::establish(path.to_str().unwrap()).unwrap();
    /// conn.batch_execute("PRAGMA journal_mode = WAL")?;
    /// conn.batch_execute("CREATE TABLE logs (line TEXT NOT NULL)")?;
    ///
    /// let outcome = conn.wal_checkpoint(None, WalCheckpointMode::Truncate)?;
    /// assert!(!outcome.busy);
    /// // The whole WAL was moved into the database file and the log truncated.
    /// assert_eq!(outcome.log_frames, Some(0));
    /// assert_eq!(outcome.checkpointed_frames, Some(0));
    /// #     Ok(())
    /// # }
    /// ```
    pub fn wal_checkpoint(
        &mut self,
        schema: Option<&str>,
        mode: WalCheckpointMode,
    ) -> QueryResult<WalCheckpointOutcome> {
        use crate::query_dsl::RunQueryDsl;

        let (busy, log_frames, checkpointed_frames) =
            WalCheckpoint { schema, mode }.get_result::<(i32, i64, i64)>(self)?;
        Ok(WalCheckpointOutcome {
            busy: busy != 0,
            // On a database not in WAL mode both counts come back as -1.
            log_frames: (log_frames >= 0).then_some(log_frames),
            checkpointed_frames: (checkpointed_frames >= 0).then_some(checkpointed_frames),
        })
    }
}

// `VACUUM` names its schema as an identifier, so that operand is quoted by the query
// builder, while the `INTO` destination is an expression and binds normally.
struct Vacuum<'a> {
    schema: Option<&'a str>,
    into: Option<&'a str>,
}

impl QueryFragment<Sqlite> for Vacuum<'_> {
    fn walk_ast<'b>(&'b self, mut out: AstPass<'_, 'b, Sqlite>) -> QueryResult<()> {
        out.push_sql("VACUUM ");
        out.push_identifier(self.schema.unwrap_or("main"))?;
        if let Some(into) = self.into {
            out.push_sql(" INTO ");
            out.push_bind_param::<crate::sql_types::Text, _>(into)?;
        }
        Ok(())
    }
}

// The schema name is runtime data, so the rendered SQL is not determined by the type.
impl QueryId for Vacuum<'_> {
    type QueryId = ();

    const HAS_STATIC_QUERY_ID: bool = false;
}

impl RunQueryDslSupport for Vacuum<'_> {}

// Like `Pragma`, no operand can be a bind parameter. Unlike `Pragma`, a
// `None` schema stays unqualified on purpose: the unqualified pragma
// checkpoints every attached database, while a qualified one targets a
// single schema. The whole checkpoint runs on the first step of the
// statement and yields exactly one row, so a prepared statement works here
// (unlike `incremental_vacuum`).
struct WalCheckpoint<'a> {
    schema: Option<&'a str>,
    mode: WalCheckpointMode,
}

impl QueryFragment<Sqlite> for WalCheckpoint<'_> {
    fn walk_ast<'b>(&'b self, mut out: AstPass<'_, 'b, Sqlite>) -> QueryResult<()> {
        out.push_sql("PRAGMA ");
        if let Some(schema) = self.schema {
            out.push_identifier(schema)?;
            out.push_sql(".");
        }
        out.push_sql(match self.mode {
            WalCheckpointMode::Passive => "wal_checkpoint(PASSIVE)",
            WalCheckpointMode::Full => "wal_checkpoint(FULL)",
            WalCheckpointMode::Restart => "wal_checkpoint(RESTART)",
            WalCheckpointMode::Truncate => "wal_checkpoint(TRUNCATE)",
            WalCheckpointMode::Noop => "wal_checkpoint(NOOP)",
        });
        Ok(())
    }
}

// The schema name and mode are runtime data, so the rendered SQL is not determined by the type.
impl QueryId for WalCheckpoint<'_> {
    type QueryId = ();

    const HAS_STATIC_QUERY_ID: bool = false;
}

impl Query for WalCheckpoint<'_> {
    type SqlType = (
        crate::sql_types::Integer,
        crate::sql_types::BigInt,
        crate::sql_types::BigInt,
    );
}

impl RunQueryDslSupport for WalCheckpoint<'_> {}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::connection::SimpleConnection;
    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    use crate::dsl::sql;
    use crate::prelude::*;
    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    use crate::result::Error;
    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    use crate::sql_types::Integer;
    use crate::sqlite::pragma::{
        AutoVacuum, FreelistCount, IncrementalVacuum, PageCount, SchemaPragma,
    };
    use core::num::NonZeroU32;

    fn connection() -> SqliteConnection {
        SqliteConnection::establish(":memory:").unwrap()
    }

    #[diesel_test_helper::test]
    fn auto_vacuum_all_modes_roundtrip_on_fresh_database() {
        for mode in [
            AutoVacuumMode::None,
            AutoVacuumMode::Full,
            AutoVacuumMode::Incremental,
        ] {
            let conn = &mut connection();
            AutoVacuum::set(mode, None, conn).unwrap();
            assert_eq!(mode, AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap());
        }
    }

    #[diesel_test_helper::test]
    fn auto_vacuum_incremental_sticks_across_schema_creation() {
        let conn = &mut connection();
        AutoVacuum::set(AutoVacuumMode::Incremental, None, conn).unwrap();
        assert_eq!(
            AutoVacuumMode::Incremental,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap()
        );

        crate::sql_query("CREATE TABLE t (id INTEGER PRIMARY KEY)")
            .execute(conn)
            .unwrap();
        assert_eq!(
            AutoVacuumMode::Incremental,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap(),
            "the mode survives once the schema exists"
        );
    }

    #[diesel_test_helper::test]
    fn auto_vacuum_change_from_none_requires_vacuum_on_populated_database() {
        let conn = &mut connection();
        crate::sql_query("CREATE TABLE t (id INTEGER PRIMARY KEY)")
            .execute(conn)
            .unwrap();
        crate::sql_query("INSERT INTO t (id) VALUES (1)")
            .execute(conn)
            .unwrap();
        assert_eq!(
            AutoVacuumMode::None,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap()
        );

        // On a populated database the switch away from `None` is silently
        // deferred until a full rewrite.
        AutoVacuum::set(AutoVacuumMode::Full, None, conn).unwrap();
        assert_eq!(
            AutoVacuumMode::None,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap(),
            "the change does not take effect without a VACUUM"
        );

        crate::sql_query("VACUUM").execute(conn).unwrap();
        assert_eq!(
            AutoVacuumMode::Full,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap(),
            "VACUUM rewrites the file and applies the mode"
        );
    }

    #[diesel_test_helper::test]
    fn auto_vacuum_targets_the_named_attached_database() {
        let conn = &mut connection();
        crate::sql_query("ATTACH DATABASE ':memory:' AS aux")
            .execute(conn)
            .unwrap();

        AutoVacuum::set(AutoVacuumMode::Full, Some("aux"), conn).unwrap();
        assert_eq!(
            AutoVacuumMode::Full,
            AutoVacuum::get::<AutoVacuumMode>(Some("aux"), conn).unwrap()
        );
        assert_eq!(
            AutoVacuumMode::None,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap(),
            "main keeps its own default"
        );
    }

    #[diesel_test_helper::test]
    fn auto_vacuum_schema_name_with_double_quote_is_handled() {
        let conn = &mut connection();
        let schema = r#"we"ird"#;
        crate::sql_query(alloc::format!(
            r#"ATTACH DATABASE ':memory:' AS "{}""#,
            schema.replace('"', "\"\"")
        ))
        .execute(conn)
        .unwrap();

        AutoVacuum::set(AutoVacuumMode::Incremental, Some(schema), conn).unwrap();
        assert_eq!(
            AutoVacuumMode::Incremental,
            AutoVacuum::get::<AutoVacuumMode>(Some(schema), conn).unwrap()
        );
    }

    table! {
        pragma_probe (id) {
            id -> Integer,
            payload -> Text,
        }
    }

    table! {
        aux.aux_pragma_probe (id) {
            id -> Integer,
            payload -> Text,
        }
    }

    const PROBE_TABLE: &str =
        "CREATE TABLE pragma_probe (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)";

    const AUX_PROBE_TABLE: &str =
        "CREATE TABLE aux.aux_pragma_probe (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)";

    // Large enough to spill onto overflow pages, so the database outgrows a single page
    // and leaves reclaimable pages behind once the row is deleted.
    fn overflowing_payload() -> String {
        "x".repeat(64 * 1024)
    }

    fn insert_overflowing_row(conn: &mut SqliteConnection) {
        crate::insert_into(pragma_probe::table)
            .values((
                pragma_probe::id.eq(1),
                pragma_probe::payload.eq(overflowing_payload()),
            ))
            .execute(conn)
            .unwrap();
    }

    #[diesel_test_helper::test]
    fn page_count_is_positive_and_grows() {
        let conn = &mut connection();
        conn.batch_execute(PROBE_TABLE).unwrap();
        let initial = PageCount::get::<i64>(None, conn).unwrap();
        assert!(initial > 0, "an initialized database has at least one page");

        insert_overflowing_row(conn);

        assert!(
            PageCount::get::<i64>(None, conn).unwrap() > initial,
            "a row spanning overflow pages grows the page count"
        );
    }

    #[diesel_test_helper::test]
    fn freelist_count_tracks_reclaimable_space() {
        let conn = &mut connection();
        assert_eq!(
            0,
            FreelistCount::get::<i64>(None, conn).unwrap(),
            "a fresh database has an empty freelist"
        );

        conn.batch_execute(PROBE_TABLE).unwrap();
        insert_overflowing_row(conn);

        crate::delete(pragma_probe::table).execute(conn).unwrap();
        assert!(
            FreelistCount::get::<i64>(None, conn).unwrap() > 0,
            "deleting the row leaves reclaimable pages on the freelist"
        );

        // `VACUUM` has no query DSL equivalent.
        crate::sql_query("VACUUM").execute(conn).unwrap();
        assert_eq!(
            0,
            FreelistCount::get::<i64>(None, conn).unwrap(),
            "VACUUM reclaims the freelist"
        );
    }

    #[diesel_test_helper::test]
    fn schema_targets_the_named_attached_database() {
        let conn = &mut connection();
        conn.batch_execute(PROBE_TABLE).unwrap();
        conn.attach_database(":memory:", "aux").unwrap();
        conn.batch_execute(AUX_PROBE_TABLE).unwrap();
        crate::insert_into(aux_pragma_probe::table)
            .values((
                aux_pragma_probe::id.eq(1),
                aux_pragma_probe::payload.eq(overflowing_payload()),
            ))
            .execute(conn)
            .unwrap();

        let main_pages = PageCount::get::<i64>(None, conn).unwrap();
        let aux_pages = PageCount::get::<i64>(Some("aux"), conn).unwrap();
        assert!(
            aux_pages > main_pages,
            "the attached database holds the data, main stays small"
        );
        assert_eq!(
            main_pages,
            PageCount::get::<i64>(Some("main"), conn).unwrap(),
            "an explicit main matches the default"
        );
    }

    #[diesel_test_helper::test]
    fn schema_name_with_backtick_is_escaped() {
        // The query builder quotes SQLite identifiers with backticks, so a backtick is
        // the character that has to be doubled.
        let conn = &mut connection();
        let schema = "back`tick";
        conn.attach_database(":memory:", schema).unwrap();
        conn.batch_execute("CREATE TABLE `back``tick`.probe (id INTEGER PRIMARY KEY)")
            .unwrap();

        assert!(PageCount::get::<i64>(Some(schema), conn).unwrap() > 0);
        assert_eq!(0, FreelistCount::get::<i64>(Some(schema), conn).unwrap());
    }

    #[cfg(not(miri))] // ffi string access
    #[diesel_test_helper::test]
    fn unknown_schema_is_reported_as_an_error() {
        let conn = &mut connection();

        assert!(PageCount::get::<i64>(Some("nope"), conn).is_err());
        assert!(FreelistCount::get::<i64>(Some("nope"), conn).is_err());
    }

    // Leaves many pages on the freelist, so `incremental_vacuum` has something to
    // reclaim and a bound smaller than the freelist is meaningful.
    fn grow_then_empty_freelist(conn: &mut SqliteConnection) {
        conn.batch_execute(PROBE_TABLE).unwrap();
        let rows = (1..=200)
            .map(|id| {
                (
                    pragma_probe::id.eq(id),
                    pragma_probe::payload.eq("x".repeat(4000)),
                )
            })
            .collect::<Vec<_>>();
        crate::insert_into(pragma_probe::table)
            .values(rows)
            .execute(conn)
            .unwrap();
        crate::delete(pragma_probe::table).execute(conn).unwrap();
    }

    // The same, in an attached schema.
    fn grow_then_empty_aux_freelist(conn: &mut SqliteConnection) {
        conn.batch_execute(AUX_PROBE_TABLE).unwrap();
        let rows = (1..=200)
            .map(|id| {
                (
                    aux_pragma_probe::id.eq(id),
                    aux_pragma_probe::payload.eq("x".repeat(4000)),
                )
            })
            .collect::<Vec<_>>();
        crate::insert_into(aux_pragma_probe::table)
            .values(rows)
            .execute(conn)
            .unwrap();
        crate::delete(aux_pragma_probe::table)
            .execute(conn)
            .unwrap();
    }

    #[diesel_test_helper::test]
    fn incremental_vacuum_clears_the_whole_freelist() {
        let conn = &mut connection();
        AutoVacuum::set(AutoVacuumMode::Incremental, None, conn).unwrap();
        grow_then_empty_freelist(conn);
        assert!(
            FreelistCount::get::<i64>(None, conn).unwrap() > 1,
            "the deleted rows should leave many pages on the freelist"
        );

        IncrementalVacuum::set(None, None, conn).unwrap();

        // Stepping the pragma only once would free a single page and leave the rest, so
        // this also pins that the statement is driven to completion.
        assert_eq!(0, FreelistCount::get::<i64>(None, conn).unwrap());
    }

    #[diesel_test_helper::test]
    fn incremental_vacuum_reclaims_at_most_the_requested_pages() {
        let conn = &mut connection();
        AutoVacuum::set(AutoVacuumMode::Incremental, None, conn).unwrap();
        grow_then_empty_freelist(conn);
        let before = FreelistCount::get::<i64>(None, conn).unwrap();
        assert!(before > 10, "the bound has to be smaller than the freelist");

        IncrementalVacuum::set(NonZeroU32::new(10), None, conn).unwrap();

        let after = FreelistCount::get::<i64>(None, conn).unwrap();
        assert!(after >= before - 10, "at most ten pages may be reclaimed");
        assert!(after < before, "some pages should have been reclaimed");
    }

    #[diesel_test_helper::test]
    fn incremental_vacuum_is_a_no_op_outside_incremental_mode() {
        let conn = &mut connection();
        assert_eq!(
            AutoVacuumMode::None,
            AutoVacuum::get::<AutoVacuumMode>(None, conn).unwrap()
        );
        grow_then_empty_freelist(conn);
        let before = FreelistCount::get::<i64>(None, conn).unwrap();
        assert!(before > 0);

        IncrementalVacuum::set(None, None, conn).unwrap();

        assert_eq!(
            before,
            FreelistCount::get::<i64>(None, conn).unwrap(),
            "a database that is not in incremental mode keeps its freelist"
        );
    }

    #[diesel_test_helper::test]
    fn incremental_vacuum_targets_the_named_attached_database() {
        let conn = &mut connection();
        conn.attach_database(":memory:", "aux").unwrap();
        AutoVacuum::set(AutoVacuumMode::Incremental, Some("aux"), conn).unwrap();

        grow_then_empty_aux_freelist(conn);
        assert!(FreelistCount::get::<i64>(Some("aux"), conn).unwrap() > 0);

        IncrementalVacuum::set(None, Some("aux"), conn).unwrap();

        assert_eq!(0, FreelistCount::get::<i64>(Some("aux"), conn).unwrap());
    }

    #[diesel_test_helper::test]
    fn incremental_vacuum_escapes_a_backtick_in_the_schema_name() {
        // An unquoted identifier would be a syntax error, and the wrong quoting would
        // address a different database.
        let conn = &mut connection();
        let schema = "back`tick";
        conn.attach_database(":memory:", schema).unwrap();

        IncrementalVacuum::set(None, Some(schema), conn).unwrap();

        assert_eq!(0, FreelistCount::get::<i64>(Some(schema), conn).unwrap());
    }

    #[cfg(not(miri))] // ffi string access
    #[diesel_test_helper::test]
    fn incremental_vacuum_of_an_unknown_schema_is_an_error() {
        let conn = &mut connection();

        assert!(IncrementalVacuum::set(None, Some("nope"), conn).is_err());
    }

    // Leaves the database holding one small row but occupying many pages, so a rebuild
    // has something to reclaim.
    fn fill_then_delete(conn: &mut SqliteConnection) {
        conn.batch_execute(PROBE_TABLE).unwrap();
        crate::insert_into(pragma_probe::table)
            .values((
                pragma_probe::id.eq(1),
                pragma_probe::payload.eq("x".repeat(256 * 1024)),
            ))
            .execute(conn)
            .unwrap();
        crate::delete(pragma_probe::table).execute(conn).unwrap();
        crate::insert_into(pragma_probe::table)
            .values((pragma_probe::id.eq(2), pragma_probe::payload.eq("kept")))
            .execute(conn)
            .unwrap();
    }

    // The same, in an attached schema.
    fn fill_then_delete_aux(conn: &mut SqliteConnection) {
        conn.batch_execute(AUX_PROBE_TABLE).unwrap();
        crate::insert_into(aux_pragma_probe::table)
            .values((
                aux_pragma_probe::id.eq(1),
                aux_pragma_probe::payload.eq("x".repeat(256 * 1024)),
            ))
            .execute(conn)
            .unwrap();
        crate::delete(aux_pragma_probe::table)
            .execute(conn)
            .unwrap();
    }

    #[diesel_test_helper::test]
    fn vacuum_repacks_the_database() {
        let conn = &mut connection();
        fill_then_delete(conn);
        let before = PageCount::get::<i64>(None, conn).unwrap();
        assert!(before > 1);

        conn.vacuum(None).unwrap();

        assert!(
            PageCount::get::<i64>(None, conn).unwrap() < before,
            "rebuilding should release the pages the deleted row occupied"
        );
        assert_eq!(
            1,
            pragma_probe::table.count().get_result::<i64>(conn).unwrap(),
            "the surviving row is still there"
        );
    }

    #[diesel_test_helper::test]
    fn vacuum_targets_the_named_attached_database() {
        let conn = &mut connection();
        conn.attach_database(":memory:", "aux").unwrap();
        fill_then_delete_aux(conn);
        let before = PageCount::get::<i64>(Some("aux"), conn).unwrap();
        assert!(before > 1);

        conn.vacuum(Some("aux")).unwrap();

        assert!(PageCount::get::<i64>(Some("aux"), conn).unwrap() < before);
    }

    #[cfg(not(miri))]
    #[diesel_test_helper::test]
    fn vacuum_inside_a_transaction_is_an_error() {
        use crate::connection::Connection;

        let conn = &mut connection();
        let result: QueryResult<()> = conn.transaction(|conn| conn.vacuum(None));

        assert!(result.is_err());
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn vacuum_into_writes_a_readable_copy_through_a_quoted_path() {
        let dir = tempfile::tempdir().unwrap();
        // A quote in the path would break a hand-assembled statement. It is a bind
        // parameter, so it is taken verbatim.
        let destination = dir.path().join("o'brien backup.db");

        let conn = &mut connection();
        conn.batch_execute(PROBE_TABLE).unwrap();
        crate::insert_into(pragma_probe::table)
            .values((pragma_probe::id.eq(1), pragma_probe::payload.eq("copied")))
            .execute(conn)
            .unwrap();

        conn.vacuum_into(None, destination.to_str().unwrap())
            .unwrap();

        let copy = &mut SqliteConnection::establish(destination.to_str().unwrap()).unwrap();
        assert_eq!(
            "copied",
            pragma_probe::table
                .select(pragma_probe::payload)
                .get_result::<String>(copy)
                .unwrap()
        );
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn vacuum_into_refuses_to_overwrite_an_existing_database() {
        let dir = tempfile::tempdir().unwrap();
        let destination = dir.path().join("occupied.db");
        {
            let occupied = &mut SqliteConnection::establish(destination.to_str().unwrap()).unwrap();
            occupied.batch_execute(PROBE_TABLE).unwrap();
        }

        let conn = &mut connection();
        conn.batch_execute(PROBE_TABLE).unwrap();

        assert!(
            conn.vacuum_into(None, destination.to_str().unwrap())
                .is_err()
        );
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn vacuum_into_copies_the_named_attached_database() {
        let dir = tempfile::tempdir().unwrap();
        let destination = dir.path().join("aux copy.db");

        let conn = &mut connection();
        conn.attach_database(":memory:", "aux").unwrap();
        conn.batch_execute(AUX_PROBE_TABLE).unwrap();
        crate::insert_into(aux_pragma_probe::table)
            .values((
                aux_pragma_probe::id.eq(7),
                aux_pragma_probe::payload.eq("copied"),
            ))
            .execute(conn)
            .unwrap();

        conn.vacuum_into(Some("aux"), destination.to_str().unwrap())
            .unwrap();

        let copy = &mut SqliteConnection::establish(destination.to_str().unwrap()).unwrap();
        // In the copy the table sits in `main`, while `aux_pragma_probe` is declared
        // schema-qualified, so this one read cannot go through it.
        let id = sql::<Integer>("SELECT id FROM aux_pragma_probe")
            .get_result::<i32>(copy)
            .unwrap();
        assert_eq!(7, id);
    }

    // no miri as that requires a fs call
    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn vacuum_into_rejects_nul_in_destination() {
        let dir = tempfile::tempdir().unwrap();
        let destination = dir.path().join("nul-truncated.db");

        let conn = &mut connection();
        assert!(matches!(
            conn.vacuum_into(None, &format!("{}\0extra.db", destination.display())),
            Err(Error::InvalidCString(_))
        ));
        assert!(!destination.exists());
    }

    #[diesel_test_helper::test]
    fn vacuum_escapes_a_backtick_in_the_schema_name() {
        let conn = &mut connection();
        let schema = "back`tick";
        conn.attach_database(":memory:", schema).unwrap();

        conn.vacuum(Some(schema)).unwrap();
    }

    #[diesel_test_helper::test]
    fn vacuuming_two_schemas_rebuilds_each_of_them() {
        // The schema is part of the rendered SQL, so the two calls must not share a
        // prepared statement. If they did, the second would rebuild the first's
        // database again and leave this one untouched.
        let conn = &mut connection();
        fill_then_delete(conn);
        conn.attach_database(":memory:", "aux").unwrap();
        fill_then_delete_aux(conn);

        let main_before = PageCount::get::<i64>(None, conn).unwrap();
        let aux_before = PageCount::get::<i64>(Some("aux"), conn).unwrap();

        conn.vacuum(None).unwrap();
        conn.vacuum(Some("aux")).unwrap();

        assert!(
            PageCount::get::<i64>(None, conn).unwrap() < main_before,
            "main was rebuilt"
        );
        assert!(
            PageCount::get::<i64>(Some("aux"), conn).unwrap() < aux_before,
            "aux was rebuilt too, not main a second time"
        );
    }

    // WAL requires a real file.
    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    fn wal_connection(path: &std::path::Path) -> SqliteConnection {
        let mut conn = SqliteConnection::establish(path.to_str().unwrap()).unwrap();
        conn.batch_execute("PRAGMA journal_mode = WAL").unwrap();
        conn
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_truncate_reports_an_emptied_wal() {
        let dir = tempfile::tempdir().unwrap();
        let conn = &mut wal_connection(&dir.path().join("wal.db"));
        conn.batch_execute(PROBE_TABLE).unwrap();
        insert_overflowing_row(conn);

        let outcome = conn
            .wal_checkpoint(None, WalCheckpointMode::Truncate)
            .unwrap();

        assert!(!outcome.busy);
        assert_eq!(Some(0), outcome.log_frames, "the WAL file was truncated");
        assert_eq!(Some(0), outcome.checkpointed_frames);
    }

    #[diesel_test_helper::test]
    fn wal_checkpoint_outside_wal_mode_reports_no_frames() {
        let conn = &mut connection();

        let outcome = conn
            .wal_checkpoint(None, WalCheckpointMode::Truncate)
            .unwrap();

        assert!(!outcome.busy);
        assert_eq!(None, outcome.log_frames);
        assert_eq!(None, outcome.checkpointed_frames);
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_accepts_every_mode() {
        let dir = tempfile::tempdir().unwrap();
        let conn = &mut wal_connection(&dir.path().join("modes.db"));
        conn.batch_execute(PROBE_TABLE).unwrap();

        for (row, mode) in [
            WalCheckpointMode::Passive,
            WalCheckpointMode::Full,
            WalCheckpointMode::Restart,
            WalCheckpointMode::Truncate,
            WalCheckpointMode::Noop,
        ]
        .into_iter()
        .enumerate()
        {
            // A fresh row per round gives every mode frames to move.
            crate::insert_into(pragma_probe::table)
                .values((
                    pragma_probe::id.eq(i32::try_from(row).unwrap() + 1),
                    pragma_probe::payload.eq("row"),
                ))
                .execute(conn)
                .unwrap();

            let outcome = conn.wal_checkpoint(None, mode).unwrap();
            assert!(!outcome.busy, "{mode:?} had no competing readers");
            assert!(
                outcome.log_frames.is_some(),
                "{mode:?} ran on a WAL database"
            );
            assert!(outcome.checkpointed_frames.is_some());
            assert!(
                outcome.checkpointed_frames <= outcome.log_frames,
                "{mode:?}: checkpointed frames cannot exceed the log size"
            );
        }
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_noop_reports_state_without_moving_frames() {
        let dir = tempfile::tempdir().unwrap();
        let conn = &mut wal_connection(&dir.path().join("noop.db"));

        // NOOP exists since SQLite 3.51.0, older versions run PASSIVE instead.
        if conn.version() < crate::connection::DatabaseVersion::new(3, 51, 0) {
            return;
        }

        conn.batch_execute(PROBE_TABLE).unwrap();
        conn.wal_checkpoint(None, WalCheckpointMode::Truncate)
            .unwrap();
        crate::insert_into(pragma_probe::table)
            .values((pragma_probe::id.eq(1), pragma_probe::payload.eq("noop")))
            .execute(conn)
            .unwrap();

        let first = conn.wal_checkpoint(None, WalCheckpointMode::Noop).unwrap();
        let second = conn.wal_checkpoint(None, WalCheckpointMode::Noop).unwrap();

        assert!(!first.busy, "NOOP never blocks");
        assert!(first.log_frames > Some(0), "the insert sits in the WAL");
        assert_eq!(Some(0), first.checkpointed_frames, "nothing was moved");
        assert_eq!(first, second, "a second NOOP reports the same state");
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_reports_busy_while_a_reader_holds_an_old_snapshot() {
        use crate::connection::Connection;

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("busy.db");
        let writer = &mut wal_connection(&path);
        writer.batch_execute(PROBE_TABLE).unwrap();
        insert_overflowing_row(writer);

        let reader = &mut SqliteConnection::establish(path.to_str().unwrap()).unwrap();
        reader
            .transaction::<_, crate::result::Error, _>(|reader| {
                // Take the read snapshot, BEGIN alone defers it to the first read.
                let _ = pragma_probe::table.count().get_result::<i64>(reader)?;

                // Grow the WAL past the reader's snapshot, so a blocking
                // checkpoint cannot complete.
                crate::insert_into(pragma_probe::table)
                    .values((pragma_probe::id.eq(2), pragma_probe::payload.eq("late")))
                    .execute(writer)?;

                // Passive is never reported busy, it checkpoints up to the
                // reader's snapshot and leaves the rest.
                let outcome = writer.wal_checkpoint(None, WalCheckpointMode::Passive)?;
                assert!(!outcome.busy, "PASSIVE never reports busy");
                assert!(
                    outcome.checkpointed_frames < outcome.log_frames,
                    "the frames past the reader's snapshot stay in the WAL"
                );

                for mode in [
                    WalCheckpointMode::Full,
                    WalCheckpointMode::Restart,
                    WalCheckpointMode::Truncate,
                ] {
                    let outcome = writer.wal_checkpoint(None, mode)?;
                    assert!(outcome.busy, "the open reader blocks a {mode:?} checkpoint");
                }
                Ok(())
            })
            .unwrap();

        let outcome = writer
            .wal_checkpoint(None, WalCheckpointMode::Truncate)
            .unwrap();
        assert!(
            !outcome.busy,
            "the checkpoint completes once the reader is done"
        );
        assert_eq!(Some(0), outcome.log_frames);
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_targets_the_named_attached_database() {
        let dir = tempfile::tempdir().unwrap();
        let conn = &mut connection();
        conn.attach_database(dir.path().join("aux.db").to_str().unwrap(), "aux")
            .unwrap();
        conn.batch_execute("PRAGMA aux.journal_mode = WAL").unwrap();
        conn.batch_execute(AUX_PROBE_TABLE).unwrap();
        crate::insert_into(aux_pragma_probe::table)
            .values((
                aux_pragma_probe::id.eq(1),
                aux_pragma_probe::payload.eq("row"),
            ))
            .execute(conn)
            .unwrap();

        let outcome = conn
            .wal_checkpoint(Some("aux"), WalCheckpointMode::Truncate)
            .unwrap();
        assert!(!outcome.busy);
        assert_eq!(
            Some(0),
            outcome.log_frames,
            "the attached database was checkpointed"
        );

        // `main` is not in WAL mode, so a checkpoint naming it reports no frames.
        let outcome = conn
            .wal_checkpoint(Some("main"), WalCheckpointMode::Truncate)
            .unwrap();
        assert_eq!(None, outcome.log_frames);
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_unqualified_covers_every_attached_database() {
        let dir = tempfile::tempdir().unwrap();
        let conn = &mut wal_connection(&dir.path().join("main.db"));
        conn.batch_execute(PROBE_TABLE).unwrap();
        insert_overflowing_row(conn);
        conn.attach_database(dir.path().join("aux.db").to_str().unwrap(), "aux")
            .unwrap();
        conn.batch_execute("PRAGMA aux.journal_mode = WAL").unwrap();
        conn.batch_execute(AUX_PROBE_TABLE).unwrap();
        crate::insert_into(aux_pragma_probe::table)
            .values((
                aux_pragma_probe::id.eq(1),
                aux_pragma_probe::payload.eq("row"),
            ))
            .execute(conn)
            .unwrap();

        conn.wal_checkpoint(None, WalCheckpointMode::Truncate)
            .unwrap();

        // Both WALs are empty afterwards, which a qualified passive
        // checkpoint reports without moving anything.
        let main_after = conn
            .wal_checkpoint(Some("main"), WalCheckpointMode::Passive)
            .unwrap();
        assert_eq!(Some(0), main_after.log_frames, "main was checkpointed");
        let aux_after = conn
            .wal_checkpoint(Some("aux"), WalCheckpointMode::Passive)
            .unwrap();
        assert_eq!(Some(0), aux_after.log_frames, "aux was checkpointed too");
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_escapes_a_double_quote_in_the_schema_name() {
        // An unquoted identifier would be a syntax error, and the wrong quoting
        // would address a different database.
        let dir = tempfile::tempdir().unwrap();
        let conn = &mut connection();
        let schema = r#"we"ird"#;
        let quoted = schema.replace('"', "\"\"");
        conn.attach_database(dir.path().join("weird.db").to_str().unwrap(), schema)
            .unwrap();
        conn.batch_execute(&alloc::format!(r#"PRAGMA "{quoted}".journal_mode = WAL"#))
            .unwrap();
        conn.batch_execute(&alloc::format!(
            r#"CREATE TABLE "{quoted}".t (id INTEGER PRIMARY KEY)"#
        ))
        .unwrap();

        let outcome = conn
            .wal_checkpoint(Some(schema), WalCheckpointMode::Truncate)
            .unwrap();
        assert_eq!(Some(0), outcome.log_frames, "the quoted schema was reached");
    }

    #[cfg(not(miri))] // ffi string access
    #[diesel_test_helper::test]
    fn wal_checkpoint_of_an_unknown_schema_is_an_error() {
        let conn = &mut connection();

        assert!(
            conn.wal_checkpoint(Some("nope"), WalCheckpointMode::Passive)
                .is_err()
        );
    }

    #[cfg(not(any(all(target_family = "wasm", target_os = "unknown"), miri)))]
    #[diesel_test_helper::test]
    fn wal_checkpoint_inside_a_transaction_is_an_error() {
        use crate::connection::Connection;

        let dir = tempfile::tempdir().unwrap();
        let conn = &mut wal_connection(&dir.path().join("txn.db"));
        conn.batch_execute(PROBE_TABLE).unwrap();

        let result: QueryResult<WalCheckpointOutcome> = conn.transaction(|conn| {
            crate::insert_into(pragma_probe::table)
                .values((pragma_probe::id.eq(1), pragma_probe::payload.eq("txn")))
                .execute(conn)?;
            conn.wal_checkpoint(None, WalCheckpointMode::Truncate)
        });

        assert!(result.is_err(), "SQLite reports SQLITE_LOCKED");
    }
}