sqll/

connection.rs

use core::ffi::CStr;
#[cfg(feature = "alloc")]
use core::ffi::c_void;
use core::ffi::{c_int, c_uint};
use core::fmt;
use core::mem::{ManuallyDrop, MaybeUninit};
use core::ops::{BitOr, Deref, DerefMut};
use core::ptr::{NonNull, null_mut};
use core::slice;

#[cfg(feature = "std")]
use std::path::Path;

use crate::ffi;
#[cfg(feature = "alloc")]
use crate::owned::Owned;
use crate::utils::{c_to_error_text, sqlite3_try};
use crate::{
    Code, DatabaseNotFound, Error, NotThreadSafe, OpenOptions, OwnedBytes, Result, Statement, Text,
};

/// A collection of flags use to prepare a statement.
pub struct Prepare(c_uint);

impl Prepare {
    /// No flags.
    ///
    /// This provides the default behavior when preparing a statement.
    pub const EMPTY: Self = Self(0);

    /// The PERSISTENT flag is a hint to the query planner that the prepared
    /// statement will be retained for a long time and probably reused many
    /// times. Without this flag, [`Connection::prepare`] assume that the
    /// prepared statement will be used just once or at most a few times and
    /// then destroyed relatively soon.
    ///
    /// The current implementation acts on this hint by avoiding the use of
    /// lookaside memory so as not to deplete the limited store of lookaside
    /// memory. Future versions of SQLite may act on this hint differently.
    pub const PERSISTENT: Self = Self(ffi::SQLITE_PREPARE_PERSISTENT as c_uint);

    /// The NORMALIZE flag is a no-op. This flag used to be required for any
    /// prepared statement that wanted to use the normalized sql interface.
    /// However, the normalized sql interface is now available to all prepared
    /// statements, regardless of whether or not they use this flag.
    pub const NORMALIZE: Self = Self(ffi::SQLITE_PREPARE_NORMALIZE as c_uint);

    /// The NO_VTAB flag causes the SQL compiler to return an error if the
    /// statement uses any virtual tables.
    pub const NO_VTAB: Self = Self(ffi::SQLITE_PREPARE_NO_VTAB as c_uint);
}

impl BitOr for Prepare {
    type Output = Self;

    fn bitor(self, rhs: Self) -> Self::Output {
        Self(self.0 | rhs.0)
    }
}

/// A SQLite database connection.
///
/// For detailed information on how to safetly use a connection, including
/// complex topics such as *Thread Safety* and asynchronous use, see
/// [`OpenOptions`].
///
/// # Examples
///
/// Opening a connection to a filesystem path:
///
/// ```no_run
/// use sqll::Connection;
///
/// let c = Connection::open("database.db")?;
///
/// c.execute(r#"
///     CREATE TABLE test (id INTEGER);
/// "#)?;
/// # Ok::<_, sqll::Error>(())
/// ```
///
/// Opening an in-memory database:
///
/// ```
/// use sqll::Connection;
///
/// let c = Connection::open_in_memory()?;
///
/// c.execute(r#"
///     CREATE TABLE test (id INTEGER);
/// "#)?;
/// # Ok::<_, sqll::Error>(())
/// ```
pub struct Connection {
    raw: NonNull<ffi::sqlite3>,
    #[cfg(feature = "alloc")]
    busy_callback: Option<Owned>,
    is_thread_safe: bool,
}

/// Connection is `Send`.
#[cfg(feature = "threadsafe")]
unsafe impl Send for Connection {}

impl Connection {
    /// Construct a connection from a raw pointer.
    #[inline]
    pub(crate) fn from_raw(raw: NonNull<ffi::sqlite3>, is_thread_safe: bool) -> Self {
        Self {
            raw,
            #[cfg(feature = "alloc")]
            busy_callback: None,
            is_thread_safe,
        }
    }

    /// Coerce this statement into a [`SendConnection`] which can be sent across
    /// threads.
    ///
    /// # Errors
    ///
    /// This return an error if neither [`full_mutex`] or [`no_mutex`] are set,
    /// or if the sqlite library is not configured to be thread safe.
    ///
    /// ```
    /// use sqll::OpenOptions;
    ///
    /// let mut c = OpenOptions::new()
    ///     .create()
    ///     .read_write()
    ///     .open_in_memory()?;
    ///
    /// let e = unsafe { c.into_send().unwrap_err() };
    /// assert!(matches!(e, sqll::NotThreadSafe { .. }));
    /// # Ok::<_, sqll::Error>(())
    /// ```
    ///
    /// [`full_mutex`]: crate::OpenOptions::full_mutex
    /// [`no_mutex`]: crate::OpenOptions::no_mutex
    ///
    /// # Safety
    ///
    /// This is unsafe because it required that the caller ensures that any
    /// database objects are synchronized. The exact level of synchronization
    /// depends on how the connection was opened:
    /// * If [`full_mutex`] was set and [`no_mutex`] was not set, no external
    ///   synchronization is necessary, but calls to the statement might block
    ///   if it's busy.
    /// * If [`no_mutex`] was set, the caller must ensure that the [`Statement`]
    ///   is fully synchronized with respect to the connection that constructed
    ///   it. One way to achieve this is to wrap all the statements behind a
    ///   single mutex.
    ///
    /// [`full_mutex`]: crate::OpenOptions::full_mutex
    /// [`no_mutex`]: crate::OpenOptions::no_mutex
    ///
    /// # Examples
    ///
    /// The following example showcases how you can share a single connection in
    /// a multi-threaded asynchronous application.
    ///
    /// > In this example, statements are compiled and executed on-the-fly. See
    /// > [`Statement::into_send`] for an example which is more idiomatic and
    /// > uses prepared statement.
    ///
    /// ```
    /// use std::sync::Arc;
    /// use sqll::{OpenOptions, SendConnection};
    /// use anyhow::Result;
    /// use tokio::task;
    /// use tokio::sync::Mutex;
    ///
    /// #[derive(Clone)]
    /// struct Database {
    ///     c: Arc<Mutex<SendConnection>>,
    /// }
    ///
    /// fn setup_database() -> Result<Database> {
    ///     let c = OpenOptions::new()
    ///         .create()
    ///         .read_write()
    ///         .no_mutex()
    ///         .open_in_memory()?;
    ///
    ///     c.execute(
    ///         r#"
    ///         CREATE TABLE users (name TEXT PRIMARY KEY NOT NULL, age INTEGER);
    ///
    ///         INSERT INTO users VALUES ('Alice', 60), ('Bob', 70), ('Charlie', 20);
    ///         "#,
    ///     )?;
    ///
    ///     // SAFETY: We serialize all accesses to the connection behind a mutex.
    ///     let c = unsafe {
    ///         c.into_send()?
    ///     };
    ///
    ///     Ok(Database {
    ///         c: Arc::new(Mutex::new(c)),
    ///     })
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<()> {
    ///     let db = setup_database()?;
    ///
    ///     let mut tasks = Vec::new();
    ///
    ///     for _ in 0..10 {
    ///         _ = task::spawn({
    ///             let db = db.clone();
    ///
    ///             async move {
    ///                 let mut c = db.c.lock_owned().await;
    ///
    ///                 let task = task::spawn_blocking(move || {
    ///                     let mut update = c.prepare("UPDATE users SET age = age + ?")?;
    ///                     update.execute(2)
    ///                 });
    ///
    ///                 Ok::<_, anyhow::Error>(task.await??)
    ///             }
    ///         });
    ///
    ///         let t = task::spawn({
    ///             let db = db.clone();
    ///
    ///             async move {
    ///                 let mut c = db.c.lock_owned().await;
    ///
    ///                 let task = task::spawn_blocking(move || -> Result<Option<i64>> {
    ///                     let mut select = c.prepare("SELECT age FROM users ORDER BY age")?;
    ///                     Ok(select.next::<i64>()?)
    ///                 });
    ///
    ///                 task.await?
    ///             }
    ///         });
    ///
    ///         tasks.push(t);
    ///     }
    ///
    ///     for t in tasks {
    ///         let first = t.await??;
    ///         assert!(matches!(first, Some(20..=40)));
    ///     }
    ///
    ///     Ok(())
    /// }
    /// ```
    pub unsafe fn into_send(self) -> Result<SendConnection, NotThreadSafe> {
        if !self.is_thread_safe {
            return Err(NotThreadSafe::connection());
        }

        Ok(SendConnection { inner: self })
    }

    /// Open a database to the given path.
    ///
    /// Note that it is possible to open an in-memory database by passing
    /// `":memory:"` here, this call might require allocating depending on the
    /// platform, so it should be avoided in favor of using [`open_in_memory`]. To avoid
    /// allocating for regular paths, you can use [`open_c_str`], however you
    /// are responsible for ensuring the c-string is a valid path.
    ///
    /// This is the same as calling:
    ///
    /// ```
    /// use sqll::OpenOptions;
    /// # let path = ":memory:";
    ///
    /// let c = OpenOptions::new()
    ///     .extended_result_codes()
    ///     .read_write()
    ///     .create()
    ///     .open(path)?;
    ///
    /// # Ok::<_, sqll::Error>(())
    /// ```
    ///
    /// [`open_in_memory`]: Self::open_in_memory
    /// [`open_c_str`]: Self::open_c_str
    #[cfg(feature = "std")]
    #[cfg_attr(docsrs, cfg(feature = "std"))]
    #[inline]
    pub fn open(path: impl AsRef<Path>) -> Result<Connection> {
        OpenOptions::new()
            .extended_result_codes()
            .read_write()
            .create()
            .open(path)
    }

    /// Open a database connection with a raw c-string.
    ///
    /// This can be used to open in-memory databases by passing `c":memory:"` or
    /// a regular open call with a filesystem path like
    /// `c"/path/to/database.sql"`.
    ///
    /// This is the same as calling:
    ///
    /// ```
    /// use sqll::OpenOptions;
    /// # let name = c":memory:";
    ///
    /// let c = OpenOptions::new()
    ///     .extended_result_codes()
    ///     .read_write()
    ///     .create()
    ///     .open_c_str(name)?;
    ///
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn open_c_str(name: &CStr) -> Result<Connection> {
        OpenOptions::new()
            .extended_result_codes()
            .read_write()
            .create()
            .open_c_str(name)
    }

    /// Open an in-memory database.
    ///
    /// This is the same as calling
    ///
    /// This is the same as calling:
    ///
    /// ```
    /// use sqll::OpenOptions;
    ///
    /// let c = OpenOptions::new()
    ///     .extended_result_codes()
    ///     .read_write()
    ///     .create()
    ///     .open_in_memory()?;
    ///
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn open_in_memory() -> Result<Connection> {
        OpenOptions::new()
            .extended_result_codes()
            .read_write()
            .create()
            .open_in_memory()
    }

    /// Check if the database connection is read-only.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Code, OpenOptions, DatabaseNotFound};
    ///
    /// let c = OpenOptions::new().read_write().open_in_memory()?;
    ///
    /// assert!(!c.database_read_only(c"main")?);
    /// let e = c.database_read_only(c"not a db").unwrap_err();
    /// assert!(matches!(e, DatabaseNotFound { .. }));
    ///
    /// let c = OpenOptions::new().read_only().open_in_memory()?;
    ///
    /// assert!(c.database_read_only(c"main")?);
    /// let e = c.database_read_only(c"not a db").unwrap_err();
    /// assert!(matches!(e, DatabaseNotFound { .. }));
    /// # Ok::<_, Box<dyn std::error::Error>>(())
    /// ```
    pub fn database_read_only(&self, name: &CStr) -> Result<bool, DatabaseNotFound> {
        unsafe {
            match ffi::sqlite3_db_readonly(self.raw.as_ptr(), name.as_ptr()) {
                1 => Ok(true),
                0 => Ok(false),
                _ => Err(DatabaseNotFound),
            }
        }
    }

    /// Execute a batch of statements.
    ///
    /// Unlike [`prepare`], this can be used to execute multiple statements
    /// separated by a semi-colon `;` and is internally optimized for one-off
    /// queries.
    ///
    /// [`prepare`]: Self::prepare
    ///
    /// # Errors
    ///
    /// If any of the statements fail, an error is returned.
    ///
    /// ```
    /// use sqll::{Code, Connection};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// let e = c.execute(":)").unwrap_err();
    /// assert_eq!(e.code(), Code::ERROR);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Result};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT, age INTEGER);
    ///
    ///     INSERT INTO users VALUES ('Alice', 42);
    ///     INSERT INTO users VALUES ('Bob', 72);
    /// "#)?;
    ///
    /// let results = c.prepare("SELECT name, age FROM users")?
    ///     .iter::<(String, u32)>()
    ///     .collect::<Result<Vec<_>>>()?;
    ///
    /// assert_eq!(results, [("Alice".to_string(), 42), ("Bob".to_string(), 72)]);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn execute(&self, stmt: impl AsRef<[u8]>) -> Result<()> {
        self.execute_raw(stmt.as_ref())
    }

    fn execute_raw(&self, stmt: &[u8]) -> Result<()> {
        unsafe {
            let mut ptr = stmt.as_ptr().cast();
            let mut len = stmt.len();

            while len > 0 {
                let mut raw = MaybeUninit::uninit();
                let mut rest = MaybeUninit::uninit();

                let l = i32::try_from(len).unwrap_or(i32::MAX);

                sqlite3_try!(
                    self,
                    ffi::sqlite3_prepare_v3(
                        self.raw.as_ptr(),
                        ptr,
                        l,
                        0,
                        raw.as_mut_ptr(),
                        rest.as_mut_ptr(),
                    )
                );

                let rest = rest.assume_init();

                // If statement is null then it's simply empty, so we can safely
                // skip it, otherwise iterate over all rows.
                if let Some(raw) = NonNull::new(raw.assume_init()) {
                    let mut statement = Statement::from_raw(raw, self.is_thread_safe);
                    while statement.step()?.is_row() {}
                }

                // Skip over empty statements.
                let o = rest.offset_from_unsigned(ptr);
                len -= o;
                ptr = rest;
            }

            Ok(())
        }
    }

    /// Enable or disable extended result codes.
    ///
    /// This can also be set during construction with
    /// [`OpenOptions::extended_result_codes`].
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{OpenOptions, Code};
    ///
    /// let mut c = OpenOptions::new().create().read_write().open_in_memory()?;
    ///
    /// let e = c.execute("
    ///     CREATE TABLE users (name TEXT);
    ///     CREATE UNIQUE INDEX idx_users_name ON users (name);
    ///
    ///     INSERT INTO users VALUES ('Bob');
    /// ");
    ///
    /// let e = c.execute("INSERT INTO users VALUES ('Bob')").unwrap_err();
    /// assert_eq!(e.code(), Code::CONSTRAINT_UNIQUE);
    /// assert_eq!(c.error_message(), "UNIQUE constraint failed: users.name");
    ///
    /// c.extended_result_codes(false)?;
    /// let e = c.execute("INSERT INTO users VALUES ('Bob')").unwrap_err();
    /// assert_eq!(e.code(), Code::CONSTRAINT);
    /// assert_eq!(c.error_message(), "UNIQUE constraint failed: users.name");
    /// # Ok::<_, sqll::Error>(())
    /// ```
    pub fn extended_result_codes(&mut self, enabled: bool) -> Result<()> {
        unsafe {
            let onoff = i32::from(enabled);
            sqlite3_try!(
                self,
                ffi::sqlite3_extended_result_codes(self.raw.as_ptr(), onoff)
            );
        }

        Ok(())
    }

    /// Get the last error message for this connection.
    ///
    /// When operating in multi-threaded environment, the error message seen
    /// here might not correspond to the query that failed unless some kind of
    /// external synchronization is in use which is the recommended way to use
    /// sqlite.
    ///
    /// This is only meaningful if an error has occured. If no errors have
    /// occured, this returns a non-erronous message like `"not an error"`
    /// (default for sqlite3).
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Code};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// let e = c.execute("
    ///     CREATE TABLE users (name TEXT);
    ///     CREATE UNIQUE INDEX idx_users_name ON users (name);
    ///
    ///     INSERT INTO users VALUES ('Bob');
    /// ");
    ///
    /// let e = c.execute("INSERT INTO users VALUES ('Bob')").unwrap_err();
    /// assert_eq!(e.code(), Code::CONSTRAINT_UNIQUE);
    /// assert_eq!(c.error_message(), "UNIQUE constraint failed: users.name");
    /// # Ok::<_, sqll::Error>(())
    /// ```
    pub fn error_message(&self) -> &Text {
        unsafe { c_to_error_text(ffi::sqlite3_errmsg(self.raw.as_ptr())) }
    }

    /// Build a prepared statement with default options.
    ///
    /// This is the same as calling `prepare_with` with without setting any
    /// options.
    ///
    /// The database connection will be kept open for the lifetime of this
    /// statement.
    ///
    /// # Errors
    ///
    /// If the prepare call contains multiple statements, it will error. To
    /// execute multiple statements, use [`execute`] instead.
    ///
    /// ```
    /// use sqll::{Connection, Code};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// let e = c.prepare("CREATE TABLE test (id INTEGER) /* test */; INSERT INTO test (id) VALUES (1);").unwrap_err();
    ///
    /// assert_eq!(e.code(), Code::MISUSE);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    ///
    /// [`execute`]: Self::execute
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE test (id INTEGER);
    /// "#)?;
    ///
    /// let mut insert_stmt = c.prepare("INSERT INTO test (id) VALUES (?);")?;
    /// let mut query_stmt = c.prepare("SELECT id FROM test;")?;
    ///
    /// drop(c);
    ///
    /// insert_stmt.execute(42)?;
    ///
    /// query_stmt.bind(())?;
    /// assert_eq!(query_stmt.iter::<i64>().collect::<Vec<_>>(), [Ok(42)]);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn prepare(&self, stmt: impl AsRef<[u8]>) -> Result<Statement> {
        self.prepare_raw(stmt.as_ref(), Prepare::EMPTY)
    }

    /// Build a prepared statement with custom flags.
    ///
    /// For long-running statements it is recommended that they have the
    /// [`Prepare::PERSISTENT`] flag set.
    ///
    /// The database connection will be kept open for the lifetime of this
    /// statement.
    ///
    /// # Errors
    ///
    /// If the prepare call contains multiple statements, it will error. To
    /// execute multiple statements, use [`execute`] instead.
    ///
    /// ```
    /// use sqll::{Connection, Code};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// let e = c.prepare_with("CREATE TABLE test (id INTEGER); INSERT INTO test (id) VALUES (1);")
    ///     .persistent()
    ///     .build()
    ///     .unwrap_err();
    ///
    /// assert_eq!(e.code(), Code::MISUSE);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    ///
    /// [`execute`]: Self::execute
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE test (id INTEGER);
    /// "#)?;
    ///
    /// let mut insert_stmt = c.prepare_with("INSERT INTO test (id) VALUES (?)")
    ///     .persistent()
    ///     .build()?;
    ///
    /// let mut query_stmt = c.prepare_with("SELECT id FROM test")
    ///     .persistent()
    ///     .build()?;
    ///
    /// drop(c);
    ///
    /// /* .. */
    ///
    /// insert_stmt.bind(42)?;
    /// assert!(insert_stmt.step()?.is_done());
    ///
    /// query_stmt.bind(())?;
    /// assert_eq!(query_stmt.iter::<i64>().collect::<Vec<_>>(), [Ok(42)]);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    pub fn prepare_with<S>(&self, stmt: S) -> PrepareWith<'_, S>
    where
        S: AsRef<[u8]>,
    {
        PrepareWith {
            conn: self,
            stmt,
            flags: Prepare::EMPTY,
        }
    }

    fn prepare_raw(&self, stmt: &[u8], flags: Prepare) -> Result<Statement> {
        unsafe {
            let mut raw = MaybeUninit::uninit();
            let mut rest = MaybeUninit::uninit();

            let ptr = stmt.as_ptr().cast();
            let len = i32::try_from(stmt.len()).unwrap_or(i32::MAX);

            sqlite3_try! {
                self,
                ffi::sqlite3_prepare_v3(
                    self.raw.as_ptr(),
                    ptr,
                    len,
                    flags.0,
                    raw.as_mut_ptr(),
                    rest.as_mut_ptr(),
                )
            };

            let rest = rest.assume_init();

            let o = rest.offset_from_unsigned(ptr);

            if o != stmt.len() {
                return Err(Error::new(
                    Code::MISUSE,
                    "multiple statements in a single prepare are not allowed",
                ));
            }

            let raw = NonNull::new_unchecked(raw.assume_init());
            Ok(Statement::from_raw(raw, self.is_thread_safe))
        }
    }

    /// Return the number of rows inserted, updated, or deleted by the most
    /// recent INSERT, UPDATE, or DELETE statement.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT, age INTEGER);
    ///
    ///     INSERT INTO users VALUES ('Alice', 42);
    ///     INSERT INTO users VALUES ('Bob', 72);
    /// "#)?;
    ///
    /// assert_eq!(c.changes(), 1);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn changes(&self) -> usize {
        unsafe { ffi::sqlite3_changes(self.raw.as_ptr()) as usize }
    }

    /// Return the total number of rows inserted, updated, and deleted by all
    /// INSERT, UPDATE, and DELETE statements since the connection was opened.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT, age INTEGER);
    ///
    ///     INSERT INTO users VALUES ('Alice', 42);
    ///     INSERT INTO users VALUES ('Bob', 72);
    /// "#)?;
    ///
    /// assert_eq!(c.total_changes(), 2);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn total_changes(&self) -> usize {
        unsafe { ffi::sqlite3_total_changes(self.raw.as_ptr()) as usize }
    }

    /// Return the rowid of the most recent successful INSERT into a rowid table
    /// or virtual table.
    ///
    /// # Examples
    ///
    /// If there is no primary key, the last inserted row id is an internal
    /// identifier for the row:
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT);
    ///
    ///     INSERT INTO users VALUES ('Alice');
    ///     INSERT INTO users VALUES ('Bob');
    /// "#)?;
    /// assert_eq!(c.last_insert_rowid(), 2);
    ///
    /// c.execute(r#"
    ///     INSERT INTO users VALUES ('Charlie');
    /// "#)?;
    /// assert_eq!(c.last_insert_rowid(), 3);
    ///
    /// let mut stmt = c.prepare("INSERT INTO users VALUES (?)")?;
    /// stmt.execute("Dave")?;
    ///
    /// assert_eq!(c.last_insert_rowid(), 4);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    ///
    /// If there is a primary key, the last inserted row id corresponds to it:
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);
    ///
    ///     INSERT INTO users (name) VALUES ('Alice');
    ///     INSERT INTO users (name) VALUES ('Bob');
    /// "#)?;
    /// assert_eq!(c.last_insert_rowid(), 2);
    ///
    /// c.execute("INSERT INTO users (name) VALUES ('Charlie')")?;
    /// assert_eq!(c.last_insert_rowid(), 3);
    ///
    /// c.execute("INSERT INTO users (name) VALUES ('Dave')")?;
    /// assert_eq!(c.last_insert_rowid(), 4);
    ///
    /// let mut select = c.prepare("SELECT id FROM users WHERE name = ?")?;
    /// select.bind("Dave")?;
    ///
    /// for id in select.iter::<i64>() {
    ///     assert_eq!(id?, 4);
    /// }
    ///
    /// c.execute("DELETE FROM users WHERE id = 3")?;
    /// assert_eq!(c.last_insert_rowid(), 4);
    ///
    /// c.execute("INSERT INTO users (name) VALUES ('Charlie')")?;
    /// assert_eq!(c.last_insert_rowid(), 5);
    ///
    /// select.bind("Charlie")?;
    ///
    /// while let Some(id) = select.next::<i64>()? {
    ///     assert_eq!(id, 5);
    /// }
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn last_insert_rowid(&self) -> i64 {
        unsafe { ffi::sqlite3_last_insert_rowid(self.raw.as_ptr()) }
    }

    /// Set a callback for handling busy events.
    ///
    /// The callback is triggered when the database cannot perform an operation
    /// due to processing of some other request. If the callback returns `true`,
    /// the operation will be repeated.
    ///
    /// The busy callback should not take any actions which modify the database
    /// connection that invoked the busy handler. In other words, the busy
    /// handler is not reentrant. Any such actions result in undefined behavior.
    ///
    /// Since this needs to allocate space to store the closure the `alloc`
    /// feature has to be enabled.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let mut c = Connection::open_in_memory()?;
    ///
    /// c.busy_handler(|attempts| {
    ///     println!("busy attempt: {attempts}");
    ///     attempts < 5
    /// })?;
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[cfg(feature = "alloc")]
    #[cfg_attr(docsrs, cfg(feature = "alloc"))]
    pub fn busy_handler<F>(&mut self, callback: F) -> Result<()>
    where
        F: FnMut(usize) -> bool + Send + 'static,
    {
        extern "C" fn glue<F>(callback: *mut c_void, attempts: c_int) -> c_int
        where
            F: FnMut(usize) -> bool,
        {
            unsafe {
                if (*(callback as *mut F))(attempts as usize) {
                    1
                } else {
                    0
                }
            }
        }

        unsafe {
            let callback = Owned::new(callback)?;

            let result = ffi::sqlite3_busy_handler(
                self.raw.as_ptr(),
                Some(glue::<F>),
                callback.as_ptr().cast(),
            );

            // NB: Old callback will be dropped and freed when we set the new
            // one here.
            self.busy_callback = Some(callback);
            sqlite3_try!(self, result);
        }

        Ok(())
    }

    /// Clear any previously registered busy handler.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let mut c = Connection::open_in_memory()?;
    ///
    /// c.busy_handler(|attempts| {
    ///     println!("busy attempt: {attempts}");
    ///     attempts < 5
    /// })?;
    ///
    /// c.clear_busy_handler()?;
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn clear_busy_handler(&mut self) -> Result<()> {
        unsafe {
            sqlite3_try! {
                self,
                ffi::sqlite3_busy_handler(
                    self.raw.as_ptr(),
                    None,
                    null_mut()
                )
            };
        }

        #[cfg(feature = "alloc")]
        {
            self.busy_callback = None;
        }

        Ok(())
    }

    /// Set an implicit callback for handling busy events that tries to repeat
    /// rejected operations until a timeout expires.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let mut c = Connection::open_in_memory()?;
    ///
    /// c.busy_timeout(5000)?;
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn busy_timeout(&mut self, ms: c_int) -> Result<()> {
        unsafe {
            sqlite3_try! {
                self,
                ffi::sqlite3_busy_timeout(
                    self.raw.as_ptr(),
                    ms
                )
            };
        }

        Ok(())
    }

    /// Serialize the database to a byte buffer.
    ///
    /// The returned buffer is allocated by sqlite and will be freed when
    /// dropped.
    ///
    /// The `name` parameter specifies the name of the database to serialize,
    /// which is typically `c"main"` for the main database.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Result};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT, age INTEGER);
    ///     INSERT INTO users VALUES ('Alice', 42);
    ///     INSERT INTO users VALUES ('Bob', 72);
    /// "#)?;
    ///
    /// let data = c.serialize(c"main")?;
    /// assert!(!data.is_empty());
    ///
    /// let c2 = Connection::open_in_memory()?;
    /// c2.deserialize(c"main", data)?;
    ///
    /// let results = c2.prepare("SELECT name, age FROM users")?
    ///     .iter::<(String, u32)>()
    ///     .collect::<Result<Vec<_>>>()?;
    ///
    /// assert_eq!(results, [("Alice".to_string(), 42), ("Bob".to_string(), 72)]);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    pub fn serialize(&self, name: &CStr) -> Result<OwnedBytes, Error> {
        unsafe {
            let mut len = MaybeUninit::uninit();

            match ffi::sqlite3_serialize(self.raw.as_ptr(), name.as_ptr(), len.as_mut_ptr(), 0) {
                ptr if ptr.is_null() => {
                    Err(Error::new(Code::NOMEM, "failed to serialize database"))
                }
                ptr => {
                    let ptr = NonNull::new_unchecked(ptr);
                    let len = len.assume_init();

                    let Ok(len) = usize::try_from(len) else {
                        ffi::sqlite3_free(ptr.as_ptr().cast());

                        return Err(Error::new(
                            Code::ERROR,
                            format_args!(
                                "failed to serialize database, returned size {len} is non-sensical"
                            ),
                        ));
                    };

                    Ok(OwnedBytes::from_raw(ptr, len))
                }
            }
        }
    }

    /// Serialize the database to a byte buffer without copying.
    ///
    /// This requires that database storage is contiguous in memory which might
    /// be difficult to satisfy. One way to do so is if the current database is
    /// [`deserialize`] from a previously serialized database (without having
    /// been modified).
    ///
    /// The `name` parameter specifies the name of the database to serialize,
    /// which is typically `c"main"` for the main database.
    ///
    /// [`deserialize`]: Self::deserialize
    ///
    /// # Safety
    ///
    /// The returned buffer is valid until the next call to `serialize` or
    /// `deserialize`, or until the connection is dropped. Modifying the
    /// database in any way might invalidate the returned buffer, so it should
    /// be used with care.
    ///
    /// For a safe variant of this method, see [`serialize`].
    ///
    /// [`serialize`]: Self::serialize
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Result};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT, age INTEGER);
    ///     INSERT INTO users VALUES ('Alice', 42);
    ///     INSERT INTO users VALUES ('Bob', 72);
    /// "#)?;
    ///
    /// let data = c.serialize(c"main")?;
    /// assert!(!data.is_empty());
    ///
    /// let c2 = Connection::open_in_memory()?;
    /// c2.deserialize(c"main", data.clone())?;
    ///
    /// let results = c2.prepare("SELECT name, age FROM users")?
    ///     .iter::<(String, u32)>()
    ///     .collect::<Result<Vec<_>>>()?;
    ///
    /// assert_eq!(results, [("Alice".to_string(), 42), ("Bob".to_string(), 72)]);
    ///
    /// let data2 = unsafe { c2.serialize_no_copy(c"main")? };
    /// assert_eq!(data, *data2);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    pub unsafe fn serialize_no_copy(&self, name: &CStr) -> Result<&[u8], Error> {
        unsafe {
            let mut len = MaybeUninit::uninit();

            match ffi::sqlite3_serialize(
                self.raw.as_ptr(),
                name.as_ptr(),
                len.as_mut_ptr(),
                ffi::SQLITE_SERIALIZE_NOCOPY as u32,
            ) {
                ptr if ptr.is_null() => Err(Error::new(
                    Code::MISUSE,
                    "database is not contiguous and cannot be serialized without copying",
                )),
                ptr => {
                    let len = len.assume_init();

                    let Ok(len) = usize::try_from(len) else {
                        return Err(Error::new(
                            Code::ERROR,
                            format_args!(
                                "failed to serialize database, returned size {len} is non-sensical"
                            ),
                        ));
                    };

                    Ok(slice::from_raw_parts(ptr.cast(), len))
                }
            }
        }
    }

    /// Deserialize and take ownership of the specified byte buffer into the
    /// database.
    ///
    /// The `name` parameter specifies the name of the database to deserialize
    /// into, which is typically `c"main"` for the main database.
    ///
    /// The `data` parameter is the byte buffer containing the serialized
    /// database, which should be obtained from a previous call to `serialize`
    /// or read from a file.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Result};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE users (name TEXT, age INTEGER);
    ///     INSERT INTO users VALUES ('Alice', 42);
    ///     INSERT INTO users VALUES ('Bob', 72);
    /// "#)?;
    ///
    /// let data = c.serialize(c"main")?;
    /// assert!(!data.is_empty());
    ///
    /// let c2 = Connection::open_in_memory()?;
    /// c2.deserialize(c"main", data)?;
    ///
    /// let results = c2.prepare("SELECT name, age FROM users")?
    ///     .iter::<(String, u32)>()
    ///     .collect::<Result<Vec<_>>>()?;
    ///
    /// assert_eq!(results, [("Alice".to_string(), 42), ("Bob".to_string(), 72)]);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    pub fn deserialize(&self, name: &CStr, data: OwnedBytes) -> Result<()> {
        let data = ManuallyDrop::new(data);

        let Ok(len) = i64::try_from(data.len()) else {
            return Err(Error::new(
                Code::MISUSE,
                "length of owned buffer is too large",
            ));
        };

        let Ok(capacity) = i64::try_from(data.capacity()) else {
            return Err(Error::new(
                Code::MISUSE,
                "capacity of owned buffer is too large",
            ));
        };

        unsafe {
            sqlite3_try! {
                self,
                ffi::sqlite3_deserialize(
                    self.raw.as_ptr(),
                    name.as_ptr(),
                    data.as_ptr().cast_mut(),
                    len,
                    capacity,
                    (ffi::SQLITE_DESERIALIZE_FREEONCLOSE | ffi::SQLITE_DESERIALIZE_RESIZEABLE) as u32,
                )
            };
        }

        Ok(())
    }
}

impl fmt::Debug for Connection {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Connection")
            .field("is_thread_safe", &self.is_thread_safe)
            .finish_non_exhaustive()
    }
}

impl Drop for Connection {
    #[inline]
    #[allow(unused_must_use)]
    fn drop(&mut self) {
        self.clear_busy_handler();

        // Will close the connection unconditionally. The database will stay
        // alive until all associated prepared statements have been closed since
        // we're using v2.
        let code = unsafe { ffi::sqlite3_close_v2(self.raw.as_ptr()) };
        debug_assert_eq!(code, ffi::SQLITE_OK);
    }
}

/// A [`Connection`] that can be sent between threads.
///
/// Constructed using [`Connection::into_send`].
pub struct SendConnection {
    inner: Connection,
}

unsafe impl Send for SendConnection {}

impl Deref for SendConnection {
    type Target = Connection;

    #[inline]
    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl DerefMut for SendConnection {
    #[inline]
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.inner
    }
}

impl fmt::Debug for SendConnection {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.inner.fmt(f)
    }
}

/// A builder for customizing a prepared [`Statement`].
///
/// See [`Connection::prepare_with`] for more details.
pub struct PrepareWith<'a, S>
where
    S: AsRef<[u8]>,
{
    conn: &'a Connection,
    stmt: S,
    flags: Prepare,
}

impl<S> PrepareWith<'_, S>
where
    S: AsRef<[u8]>,
{
    /// Set custom flags for the prepared statement.
    ///
    /// When setting [`Prepare::PERSISTENT`] it is recommended to use
    /// [`PrepareWith::persistent`] instead for improved readability.
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::{Connection, Prepare};
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE test (id INTEGER);
    /// "#)?;
    ///
    /// let mut stmt = c.prepare_with("SELECT id FROM test")
    ///     .with_flags(Prepare::PERSISTENT | Prepare::NO_VTAB)
    ///     .build()?;
    ///
    /// stmt.bind(())?;
    /// assert_eq!(stmt.iter::<i64>().collect::<Vec<_>>(), []);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn with_flags(mut self, flags: Prepare) -> Self {
        self.flags = flags;
        self
    }

    /// Set the [`Prepare::PERSISTENT`] flag for the built prepared
    /// [`Statement`].
    ///
    /// # Examples
    ///
    /// ```
    /// use sqll::Connection;
    ///
    /// let c = Connection::open_in_memory()?;
    ///
    /// c.execute(r#"
    ///     CREATE TABLE test (id INTEGER);
    /// "#)?;
    ///
    /// let mut insert_stmt = c.prepare_with("INSERT INTO test (id) VALUES (?)")
    ///     .persistent()
    ///     .build()?;
    ///
    /// let mut query_stmt = c.prepare_with("SELECT id FROM test")
    ///     .persistent()
    ///     .build()?;
    ///
    /// drop(c);
    ///
    /// /* .. */
    ///
    /// insert_stmt.bind(42)?;
    /// assert!(insert_stmt.step()?.is_done());
    ///
    /// query_stmt.bind(())?;
    /// assert_eq!(query_stmt.iter::<i64>().collect::<Vec<_>>(), [Ok(42)]);
    /// # Ok::<_, sqll::Error>(())
    /// ```
    #[inline]
    pub fn persistent(self) -> Self {
        self.with_flags(Prepare::PERSISTENT)
    }

    /// Build the prepared statement.
    #[inline]
    pub fn build(self) -> Result<Statement> {
        self.conn.prepare_raw(self.stmt.as_ref(), self.flags)
    }
}