liteboxfs 0.2.0

use std::{path::Path, time::SystemTime};

use rusqlite::{ToSql, functions::FunctionFlags, types::FromSql};
#[cfg(feature = "_encryption")]
use secrecy::{SecretSlice, SecretString};
use uuid::Uuid;

#[cfg(all(feature = "fs", target_os = "linux"))]
use crate::lock::LockHandle;
use crate::{
    DEFAULT_ROOT_NAME, Filesystem, FilesystemId, RootId, Transaction, TransactionBehavior,
    errors::InternalError,
    lock::{LockHandler, LockResult, LockType},
    settings::{
        Chunking, Compression, FormatVersion, SQLITE_APPLICATION_ID, Settings, SettingsValidator,
    },
    sql::SqlStore,
    util::system_time_to_nanos,
};
#[cfg(all(feature = "fuse", target_os = "linux"))]
use crate::{MountOption, RootBy};

const MIGRATION_V1: &str = include_str!("migrations/v001_schema.sql");

// The number of prepared statements rusqlite caches out of the box is fairly small.
const PREPARED_STATEMENTS_CACHE_SIZE: usize = 64;

#[cfg(feature = "_encryption")]
#[derive(Debug, Clone)]
enum EncryptionMode {
    Password(SecretString),
    Key(SecretSlice<u8>),
}

/// A secret (password or key) used to encrypt a litebox via SQLCipher.
#[cfg(feature = "_encryption")]
#[derive(Debug, Clone)]
pub struct EncryptionSecret {
    inner: EncryptionMode,
}

#[cfg(feature = "_encryption")]
impl EncryptionSecret {
    /// The length of a SQLCipher encryption key in bytes.
    pub const KEY_LEN: usize = 32;

    /// Create a new [`EncryptionSecret`] from a password.
    #[allow(unused_variables)]
    pub fn password<S: AsRef<str> + ?Sized>(password: &S) -> Self {
        {
            Self {
                inner: EncryptionMode::Password(password.as_ref().into()),
            }
        }
    }

    /// Create a new [`EncryptionSecret`] from a fixed-length key.
    #[allow(unused_variables)]
    pub fn key(key: &[u8; Self::KEY_LEN]) -> Self {
        {
            Self {
                inner: EncryptionMode::Key(key.to_vec().into()),
            }
        }
    }
}

/// A builder for configuring a new litebox.
#[derive(Debug, Clone)]
pub struct CreateOptions {
    // We name these something different from the corresponding methods because if a user tries to
    // call one of the methods without the corresponding feature enabled, the compiler will tell
    // them "private field, not a method" (E0599), which is confusing.
    enable_compression: bool,
    enable_chunking: bool,
    custom_block_size: Option<usize>,
    #[cfg(feature = "_encryption")]
    secret: Option<EncryptionMode>,
    skip_application_id: bool,
}

impl Default for CreateOptions {
    fn default() -> Self {
        Self::new()
    }
}

impl CreateOptions {
    const MIN_BLOCK_SIZE: usize = 1024 * 16;
    const MAX_BLOCK_SIZE: usize = 1024 * 128;

    /// Create a new [`CreateOptions`] with default settings.
    pub fn new() -> Self {
        Self {
            enable_compression: false,
            enable_chunking: false,
            custom_block_size: None,
            #[cfg(feature = "_encryption")]
            secret: None,
            skip_application_id: false,
        }
    }

    /// Enable or disable file compression.
    ///
    /// This compresses data to reduce storage usage at the cost of performance.
    ///
    /// The default value is `false`.
    ///
    /// # See Also
    ///
    /// - [`Filesystem::compression_enabled`]
    /// - [`Filesystem::set_compression`]
    #[cfg(feature = "compression")]
    pub fn compression(mut self, enabled: bool) -> Self {
        self.enable_compression = enabled;
        self
    }

    /// Enable or disable content-defined chunking.
    ///
    /// This deduplicates data at the sub-file level to reduce storage usage for similar files at
    /// the cost of performance.
    ///
    /// The default value is `false`.
    ///
    /// # See Also
    ///
    /// - [`Filesystem::chunking_enabled`]
    /// - [`Filesystem::set_chunking`]
    #[cfg(feature = "chunking")]
    pub fn chunking(mut self, enabled: bool) -> Self {
        self.enable_chunking = enabled;
        self
    }

    /// Encrypt the litebox with the given secret via SQLCipher.
    #[cfg(feature = "_encryption")]
    pub fn encryption_secret(mut self, secret: EncryptionSecret) -> Self {
        self.secret = Some(secret.inner);
        self
    }

    /// Change the block size from its default.
    ///
    /// The block size has implications for performance, storage efficiency, and memory usage. The
    /// correct value depends on your workload.
    ///
    /// The block size must be a power of 2 between 16 KiB and 128 KiB, however these limits may
    /// change in future versions. A value outside this range will be clamped to the nearest bound.
    /// A value that is not a power of 2 will be rounded down to the nearest power of 2.
    ///
    /// The default block size is currently 32 KiB, however this may change in future versions.
    pub fn block_size(mut self, bytes: usize) -> Self {
        self.custom_block_size = Some(bytes);
        self
    }

    /// Skip setting the SQLite `application_id`.
    ///
    /// By default, creating a litebox sets the SQLite
    /// [`application_id`](https://sqlite.org/pragma.html#pragma_application_id) to the magic
    /// number `0x6c626f78` (`b"lbox"`). If you are using LiteboxFS to implement your own
    /// application file format, you may want to set the `application_id` yourself.
    ///
    /// The default value is `false`.
    ///
    /// # See Also
    ///
    /// - [`OpenOptions::ignore_application_id`]
    pub fn skip_application_id(mut self, skip: bool) -> Self {
        self.skip_application_id = skip;
        self
    }
}

impl TryFrom<CreateOptions> for Settings {
    type Error = crate::Error;

    fn try_from(options: CreateOptions) -> Result<Self, Self::Error> {
        if options.enable_compression && cfg!(not(feature = "compression")) {
            return Err(InternalError::CompressionDisabled.into());
        }

        if options.enable_chunking && cfg!(not(feature = "chunking")) {
            return Err(InternalError::ChunkingDisabled.into());
        }

        let chunk_size = 2usize.pow(
            options
                .custom_block_size
                .unwrap_or(32 * 1024)
                .clamp(CreateOptions::MIN_BLOCK_SIZE, CreateOptions::MAX_BLOCK_SIZE)
                .ilog2(),
        );

        Ok(Settings {
            uuid: FilesystemId::new(),
            version: FormatVersion::CURRENT,
            compression: if options.enable_compression {
                Compression::ZSTD
            } else {
                Compression::None
            },
            chunking: if options.enable_chunking {
                Chunking::new_cdc(chunk_size)
            } else {
                Chunking::Fixed { size: chunk_size }
            },
            logical_block_size: chunk_size * 2,
            merkle_branch_factor: 16,
            small_write_threshold: chunk_size / 16,
            min_write_buffer_size: 1024 * 512,
        })
    }
}

/// A builder for options when opening a litebox.
#[derive(Debug, Clone)]
pub struct OpenOptions {
    readonly: bool,
    #[cfg(feature = "_encryption")]
    secret: Option<EncryptionMode>,
    ignore_application_id: bool,
}

impl Default for OpenOptions {
    fn default() -> Self {
        Self::new()
    }
}

impl OpenOptions {
    /// Create a new [`OpenOptions`] with default settings.
    pub fn new() -> Self {
        Self {
            readonly: false,
            #[cfg(feature = "_encryption")]
            secret: None,
            ignore_application_id: false,
        }
    }

    /// Open the database in read-only mode.
    ///
    /// The default value is `false`.
    pub fn readonly(mut self, readonly: bool) -> Self {
        self.readonly = readonly;
        self
    }

    /// The SQLCipher encryption secret for the litebox.
    #[cfg(feature = "_encryption")]
    pub fn encryption_secret(mut self, secret: EncryptionSecret) -> Self {
        self.secret = Some(secret.inner);
        self
    }

    /// Ignore the SQLite `application_id` when opening a litebox.
    ///
    /// By default, opening a litebox checks for the magic number `0x6c626f78` (`b"lbox"`) in the
    /// SQLite [`application_id`](https://sqlite.org/pragma.html#pragma_application_id) and returns
    /// an error if it doesn't match.
    ///
    /// The default value is `false`.
    ///
    /// # See Also
    ///
    /// - [`CreateOptions::skip_application_id`]
    pub fn ignore_application_id(mut self, ignore: bool) -> Self {
        self.ignore_application_id = ignore;
        self
    }
}

/// The SQLite journal mode.
///
/// You can change the journal mode of a connection with [`Connection::set_journal_mode`].
///
/// See the [SQLite documentation](https://sqlite.org/pragma.html#pragma_journal_mode) on journal
/// modes for details.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum JournalMode {
    /// The `DELETE` journal mode.
    ///
    /// This is the default SQLite journal mode.
    Delete,

    /// The `WAL` journal mode.
    ///
    /// Write-ahead logging can improve performance and concurrency, but comes with caveats. See
    /// the [SQLite documentation](https://sqlite.org/wal.html) on write-ahead logging for details.
    ///
    /// Bench testing has found that enabling write-ahead logging for LiteboxFS improves read and
    /// write performance for small files, but degrades performance for large files. Your results
    /// may vary, so if performance is critical, consider benchmarking your workload to see whether
    /// enabling write-ahead logging makes sense.
    Wal,
}

#[doc(hidden)]
impl ToSql for JournalMode {
    fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
        Ok(rusqlite::types::ToSqlOutput::from(match self {
            JournalMode::Delete => "DELETE",
            JournalMode::Wal => "WAL",
        }))
    }
}

#[doc(hidden)]
impl FromSql for JournalMode {
    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
        match value.as_str()?.to_ascii_uppercase().as_str() {
            "DELETE" => Ok(JournalMode::Delete),
            "WAL" => Ok(JournalMode::Wal),
            _ => Err(rusqlite::types::FromSqlError::InvalidType),
        }
    }
}

#[derive(Debug)]
enum LiteboxLock {
    #[cfg(all(feature = "fs", target_os = "linux"))]
    InMemory,
    #[cfg(all(feature = "fs", target_os = "linux"))]
    #[cfg_attr(not(all(feature = "fuse", target_os = "linux")), allow(dead_code))]
    OnDisk(LockHandle),
    #[cfg(all(feature = "fuse", target_os = "linux"))]
    Mounted,
    #[cfg(not(all(feature = "fs", target_os = "linux")))]
    Disabled,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ShouldAcquireLock {
    Yes,
    No,
}

/// A connection to a litebox.
///
/// You can open a connection to a litebox with one of these methods:
///
/// - [`Connection::open`]
/// - [`Connection::create`]
/// - [`Connection::create_new`]
/// - [`Connection::open_in_memory`]
///
/// From there, all filesystem operations happen within a transaction. You can start a new
/// transaction with [`Connection::tx`] or [`Connection::tx_with`]. Most of the time, you'll use
/// [`Connection::exec`] or [`Connection::exec_with`] instead, which execute a closure within a
/// transaction, committing if it returns [`Ok`] and rolling back if it returns [`Err`].
///
/// Within a transaction, you can use [`Filesystem`] to perform filesystem operations.
///
/// When opening a connection, consider whether you want to enable [write-ahead
/// logging](https://sqlite.org/wal.html) using [`Connection::set_journal_mode`].
#[derive(Debug)]
pub struct Connection {
    conn: rusqlite::Connection,
    settings: Settings,
    default_root_id: RootId,
    // Held for the lifetime of the connection to a file-backed litebox to keep a FUSE mount from
    // running concurrently with other connections. In-memory connections don't need locking.
    #[cfg_attr(not(all(feature = "fuse", target_os = "linux")), allow(dead_code))]
    litebox_lock: LiteboxLock,
}

impl Connection {
    fn new(mut conn: rusqlite::Connection, lock: ShouldAcquireLock) -> crate::Result<Self> {
        conn.pragma_update(None, "foreign_keys", "ON")?;
        conn.set_prepared_statement_cache_capacity(PREPARED_STATEMENTS_CACHE_SIZE);

        let mut stmt = conn.prepare(
            r#"
            SELECT
                key,
                value
            FROM
                liteboxfs_settings;
            "#,
        )?;

        let settings = Settings::try_from(SettingsValidator::from_iter(
            stmt.query_map([], |row| {
                let key: String = row.get(0)?;
                let value: String = row.get(1)?;
                Ok((key, value))
            })?
            .collect::<Result<Vec<_>, _>>()?,
        ))?;

        drop(stmt);

        #[cfg(not(feature = "compression"))]
        {
            let has_encoded: bool = conn
                .query_row(
                    "SELECT 1 FROM liteboxfs_blocks WHERE encoding != 0 LIMIT 1",
                    [],
                    |_| Ok(true),
                )
                .unwrap_or(false);

            if has_encoded {
                return Err(InternalError::CompressionDisabled.into());
            }
        }

        let default_root_id: RootId = conn.query_row(
            "SELECT uuid FROM liteboxfs_roots WHERE name = ?1",
            [DEFAULT_ROOT_NAME],
            |row| {
                let uuid_str: String = row.get(0)?;
                uuid_str.parse::<RootId>().map_err(|e| {
                    rusqlite::Error::FromSqlConversionFailure(
                        0,
                        rusqlite::types::Type::Text,
                        Box::new(e),
                    )
                })
            },
        )?;

        // We acquire the litebox lock before garbage collection so that a mount holding the
        // exclusive lock can rely on no other connection mutating the database, even via the GC
        // pass that runs on connection open. The lock can only be enforced when the `fs` feature
        // is on and we're on Linux; on other targets we silently skip it.
        #[cfg(all(feature = "fs", target_os = "linux"))]
        let litebox_lock = if lock == ShouldAcquireLock::Yes {
            let handler = LockHandler::new(settings.uuid);
            match handler.acquire_litebox_lock(LockType::Read)? {
                LockResult::Acquired(handle) => LiteboxLock::OnDisk(handle),
                LockResult::Locked => return Err(crate::Error::LiteboxLocked),
            }
        } else {
            LiteboxLock::InMemory
        };
        #[cfg(not(all(feature = "fs", target_os = "linux")))]
        let litebox_lock = {
            let _ = lock;
            LiteboxLock::Disabled
        };

        {
            let mut tx = conn.transaction()?;
            Self::garbage_collect_unlinked_files(&mut tx, settings.clone())?;
            tx.commit()?;
        }

        Ok(Self {
            conn,
            settings,
            default_root_id,
            litebox_lock,
        })
    }

    #[cfg(feature = "_encryption")]
    fn key_to_hex_pragma(key: &[u8]) -> String {
        use std::fmt::Write;

        let mut s = String::with_capacity(4 + key.len() * 2);

        s.push_str("x'");
        for byte in key {
            write!(s, "{:02x}", byte).unwrap();
        }
        s.push('\'');

        s
    }

    #[cfg(feature = "_encryption")]
    #[allow(unused_variables)]
    fn init_encryption(
        conn: &mut rusqlite::Connection,
        secret: &EncryptionMode,
    ) -> crate::Result<()> {
        match secret {
            EncryptionMode::Password(password) => {
                use secrecy::ExposeSecret;
                conn.pragma_update(None, "key", password.expose_secret())?;
                Ok(())
            }
            EncryptionMode::Key(key) => {
                use secrecy::ExposeSecret;
                conn.pragma_update(None, "key", Self::key_to_hex_pragma(key.expose_secret()))?;
                Ok(())
            }
        }
    }

    fn is_litebox(conn: &rusqlite::Connection) -> crate::Result<bool> {
        let mut is_litebox = false;

        conn.pragma_query(None, "application_id", |row| {
            let application_id: i32 = row.get(0)?;
            is_litebox = application_id == SQLITE_APPLICATION_ID;
            Ok(())
        })?;

        Ok(is_litebox)
    }

    fn init_settings(tx: &rusqlite::Transaction, settings: &Settings) -> crate::Result<()> {
        let mut stmt = tx.prepare(
            r#"
                INSERT INTO
                    liteboxfs_settings (key, value)
                VALUES
                    (?, ?);
                "#,
        )?;

        for (key, value) in settings {
            stmt.execute(rusqlite::params![key, value])?;
        }

        Ok(())
    }

    fn init_functions(conn: &mut rusqlite::Connection) -> crate::Result<()> {
        conn.create_scalar_function(
            "now_nanos_blob",
            0,
            FunctionFlags::SQLITE_INNOCUOUS,
            move |ctx| {
                assert_eq!(
                    ctx.len(),
                    0,
                    "Called with arguments when none were expected."
                );

                system_time_to_nanos(SystemTime::now())
                    .map_err(|err| rusqlite::Error::UserFunctionError(Box::new(err)))
            },
        )?;

        conn.create_scalar_function(
            "new_uuid_v4",
            0,
            FunctionFlags::SQLITE_INNOCUOUS,
            move |ctx| {
                assert_eq!(
                    ctx.len(),
                    0,
                    "Called with arguments when none were expected."
                );

                Ok(Uuid::new_v4().to_string())
            },
        )?;

        Ok(())
    }

    // Typically, unlinked files are deleted when their last file handle is dropped. However, this
    // relies on the `Drop` implementation, which may not be called if the program crashes. In this
    // case, we would end up with an orphaned file in the database forever. The same thing could
    // happen if deleting the file in the drop handler fails for any reason. This will always
    // happen if the user acquires a raw file handle and neglects to manually release it.
    //
    // To safeguard against this, we do a one-time garbage collection when we open a connection to
    // the database. Just like when dropping a file handle, we only delete unlinked files that are
    // not currently open in any other transactions, which we enforce using file descriptor locks.
    fn garbage_collect_unlinked_files(
        tx: &mut rusqlite::Transaction,
        settings: Settings,
    ) -> crate::Result<()> {
        let savepoint = tx.savepoint()?;
        let lock_handler = LockHandler::new(settings.uuid);
        let store = SqlStore::new(savepoint, settings);

        let unlinked_file_ids = store.list_unlinked_files()?;

        for file_id in unlinked_file_ids {
            match lock_handler.acquire_file_lock(file_id, LockType::Write)? {
                LockResult::Acquired(_lock_handle) => {
                    store.delete_file_if_unlinked(file_id)?;
                }
                LockResult::Locked => {}
            }
        }

        store.commit()?;

        Ok(())
    }

    fn init(
        conn: &mut rusqlite::Connection,
        settings: &Settings,
        skip_application_id: bool,
    ) -> crate::Result<()> {
        Self::init_functions(conn)?;

        let tx = conn.transaction()?;

        tx.execute_batch(MIGRATION_V1)?;

        if !skip_application_id {
            tx.pragma_update(
                None,
                "application_id",
                crate::settings::SQLITE_APPLICATION_ID,
            )?;
        }

        Self::init_settings(&tx, settings)?;

        tx.commit()?;

        Ok(())
    }

    /// Open a connection to the litebox at `path`.
    ///
    /// This does not create a new litebox if one does not already exist.
    ///
    /// # Errors
    ///
    /// - [`CannotOpen`]: The litebox could not be opened for some reason, such as because it does
    ///   not exist or the current user does not have permission.
    /// - [`NotADatabase`]: The file at `path` is not a SQLite database or could not be decrypted.
    /// - [`NotALitebox`]: The SQLite database at `path` is not a litebox.
    /// - [`FeatureDisabled`]: The litebox at `path` uses a feature that is not enabled in this
    ///   build of LiteboxFS.
    /// - [`LiteboxLocked`]: The litebox is currently mounted by another process.
    ///
    /// [`CannotOpen`]: crate::Error::CannotOpen
    /// [`NotADatabase`]: crate::Error::NotADatabase
    /// [`NotALitebox`]: crate::Error::NotALitebox
    /// [`FeatureDisabled`]: crate::Error::FeatureDisabled
    /// [`LiteboxLocked`]: crate::Error::LiteboxLocked
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use liteboxfs::{Connection, OpenOptions};
    /// let opts = OpenOptions::new();
    /// let conn = Connection::open("example.litebox", &opts)?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn open<P: AsRef<Path>>(path: P, options: &OpenOptions) -> crate::Result<Self> {
        use rusqlite::OpenFlags;

        // SQLITE_OPEN_NO_MUTEX is the default in rusqlite. Its docs explain why.
        let flags = OpenFlags::SQLITE_OPEN_NO_MUTEX
            | if options.readonly {
                OpenFlags::SQLITE_OPEN_READ_ONLY
            } else {
                OpenFlags::SQLITE_OPEN_READ_WRITE
            };

        let mut conn = rusqlite::Connection::open_with_flags(path, flags)?;

        #[cfg(feature = "_encryption")]
        if let Some(secret) = &options.secret {
            Self::init_encryption(&mut conn, secret)?;
        }

        if !options.ignore_application_id && !Self::is_litebox(&conn)? {
            return Err(crate::Error::NotALitebox);
        }

        Self::init_functions(&mut conn)?;

        Self::new(conn, ShouldAcquireLock::Yes)
    }

    /// Create or open the litebox at `path`.
    ///
    /// This creates the litebox if it does not already exist.
    ///
    /// # Errors
    ///
    /// - [`CannotOpen`]: The litebox could not be opened for some reason, such as because the
    ///   current user does not have permission.
    /// - [`NotADatabase`]: The file at `path` exists but is not a SQLite database.
    /// - [`FeatureDisabled`]: The litebox at `path` uses a feature that is not enabled in this
    ///   build of LiteboxFS.
    /// - [`LiteboxLocked`]: The litebox is currently mounted by another process.
    ///
    /// [`CannotOpen`]: crate::Error::CannotOpen
    /// [`NotADatabase`]: crate::Error::NotADatabase
    /// [`FeatureDisabled`]: crate::Error::FeatureDisabled
    /// [`LiteboxLocked`]: crate::Error::LiteboxLocked
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use liteboxfs::{Connection, CreateOptions};
    /// let opts = CreateOptions::new();
    /// let conn = Connection::create("example.litebox", &opts)?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn create<P: AsRef<Path>>(path: P, options: &CreateOptions) -> crate::Result<Self> {
        use rusqlite::OpenFlags;

        // SQLITE_OPEN_NO_MUTEX is the default in rusqlite. Its docs explain why.
        let flags = OpenFlags::SQLITE_OPEN_NO_MUTEX
            | OpenFlags::SQLITE_OPEN_READ_WRITE
            | OpenFlags::SQLITE_OPEN_CREATE;

        let mut conn = rusqlite::Connection::open_with_flags(path, flags)?;

        #[cfg(feature = "_encryption")]
        if let Some(secret) = &options.secret {
            Self::init_encryption(&mut conn, secret)?;
        }

        if Self::is_litebox(&conn)? {
            Self::init_functions(&mut conn)?;
        } else {
            Self::init(
                &mut conn,
                &options.to_owned().try_into()?,
                options.skip_application_id,
            )?;
        }

        Self::new(conn, ShouldAcquireLock::Yes)
    }

    /// Create a new litebox at `path`.
    ///
    /// This fails if a file already exists at `path`.
    ///
    /// # Errors
    ///
    /// - [`NotADatabase`]: The file at `path` already exists and is not a SQLite database.
    /// - [`LiteboxAlreadyExists`]: A litebox already exists at `path`.
    /// - [`LiteboxLocked`]: The litebox is currently mounted by another process.
    ///
    /// [`NotADatabase`]: crate::Error::NotADatabase
    /// [`LiteboxAlreadyExists`]: crate::Error::LiteboxAlreadyExists
    /// [`LiteboxLocked`]: crate::Error::LiteboxLocked
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use liteboxfs::{Connection, CreateOptions};
    /// let opts = CreateOptions::new();
    /// let conn = Connection::create_new("example.litebox", &opts)?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn create_new<P: AsRef<Path>>(path: P, options: &CreateOptions) -> crate::Result<Self> {
        use rusqlite::OpenFlags;

        // SQLITE_OPEN_NO_MUTEX is the default in rusqlite. Its docs explain why.
        let flags = OpenFlags::SQLITE_OPEN_NO_MUTEX
            | OpenFlags::SQLITE_OPEN_READ_WRITE
            | OpenFlags::SQLITE_OPEN_CREATE;

        let mut conn = rusqlite::Connection::open_with_flags(path, flags)?;

        #[cfg(feature = "_encryption")]
        if let Some(secret) = &options.secret {
            Self::init_encryption(&mut conn, secret)?;
        }

        if Self::is_litebox(&conn)? {
            return Err(crate::Error::LiteboxAlreadyExists);
        }

        Self::init(
            &mut conn,
            &options.to_owned().try_into()?,
            options.skip_application_id,
        )?;

        Self::new(conn, ShouldAcquireLock::Yes)
    }

    /// Create a new in-memory litebox.
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Connection, CreateOptions};
    /// let opts = CreateOptions::new();
    /// let conn = Connection::open_in_memory(&opts)?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn open_in_memory(options: &CreateOptions) -> crate::Result<Self> {
        let mut conn = rusqlite::Connection::open_in_memory()?;

        #[cfg(feature = "_encryption")]
        if let Some(secret) = &options.secret {
            Self::init_encryption(&mut conn, secret)?;
        }

        Self::init(
            &mut conn,
            &options.to_owned().try_into()?,
            options.skip_application_id,
        )?;

        Self::new(conn, ShouldAcquireLock::No)
    }

    /// Create a new in-memory litebox.
    #[cfg(test)]
    pub(crate) fn open_for_testing(settings: &Settings) -> crate::Result<Self> {
        let mut conn = rusqlite::Connection::open_in_memory()?;

        Self::init(&mut conn, settings, false)?;

        Self::new(conn, ShouldAcquireLock::No)
    }

    // Because test FUSE implementation is tested using external tools rather than via the unit
    // test suite, we disable test coverage reporting for methods that are only used by the FUSE
    // adapter.

    /// The total number of database rows that have changed since this connection was opened.
    ///
    /// This is used by the FUSE adapter to detect whether a long-lived transaction has accumulated
    /// any changes.
    #[cfg(feature = "fuse")]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub(crate) fn total_changes(&self) -> u64 {
        self.conn.total_changes()
    }

    /// Get the settings of this connection.
    #[cfg(feature = "fuse")]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub(crate) fn settings(&self) -> &Settings {
        &self.settings
    }

    /// Get the default root ID of this connection.
    #[cfg(feature = "fuse")]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub(crate) fn default_root_id(&self) -> RootId {
        self.default_root_id
    }

    /// Execute a single SQL statement on the underlying SQLite connection.
    ///
    /// This is used by the FUSE adapter to issue raw `BEGIN` / `COMMIT` against a long-lived
    /// transaction.
    #[cfg(feature = "fuse")]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub(crate) fn execute_batch(&mut self, sql: &str) -> crate::Result<()> {
        self.conn.execute_batch(sql)?;
        Ok(())
    }

    /// Open a savepoint directly on the underlying SQLite connection.
    ///
    /// This is used by the FUSE adapter to derive a [`Filesystem`] from a long-lived transaction
    /// managed via raw `BEGIN` / `COMMIT` SQL rather than a [`rusqlite::Transaction`] wrapper.
    #[cfg(feature = "fuse")]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub(crate) fn raw_savepoint(&mut self) -> crate::Result<rusqlite::Savepoint<'_>> {
        Ok(self.conn.savepoint()?)
    }

    /// Delete orphaned blocks. Mirrors [`Transaction::clean`] but operates directly on the
    /// connection so that the FUSE adapter can run it before a manual `COMMIT`.
    #[cfg(feature = "fuse")]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub(crate) fn clean_orphan_blocks(&mut self) -> crate::Result<()> {
        let mut stmt = self.conn.prepare_cached(
            r#"
            DELETE FROM
                liteboxfs_blocks
            WHERE
                id NOT IN (SELECT block FROM liteboxfs_file_blocks);
            "#,
        )?;

        stmt.execute([])?;

        Ok(())
    }

    /// Start a new transaction.
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Connection, CreateOptions, FileKind, Owner};
    /// # let opts = CreateOptions::new();
    /// # let mut conn = Connection::open_in_memory(&opts)?;
    /// let mut tx = conn.tx()?;
    /// {
    ///     let mut fs = tx.fs()?;
    ///     fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// }
    /// tx.commit()?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn tx<'conn>(&'conn mut self) -> crate::Result<Transaction<'conn>> {
        Ok(Transaction::new(
            self.conn.transaction()?,
            self.settings.clone(),
            self.default_root_id,
        ))
    }

    /// Start a new transaction with the given [`TransactionBehavior`].
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Connection, CreateOptions, FileKind, Owner, TransactionBehavior};
    /// # let opts = CreateOptions::new();
    /// # let mut conn = Connection::open_in_memory(&opts)?;
    /// let mut tx = conn.tx_with(TransactionBehavior::Immediate)?;
    /// {
    ///     let mut fs = tx.fs()?;
    ///     fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// }
    /// tx.commit()?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn tx_with<'conn>(
        &'conn mut self,
        behavior: TransactionBehavior,
    ) -> crate::Result<Transaction<'conn>> {
        Ok(Transaction::new(
            self.conn.transaction_with_behavior(behavior.into())?,
            self.settings.clone(),
            self.default_root_id,
        ))
    }

    /// Execute the given function within a new transaction.
    ///
    /// See [`Transaction::exec`].
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Connection, CreateOptions, FileKind, Owner};
    /// # let opts = CreateOptions::new();
    /// # let mut conn = Connection::open_in_memory(&opts)?;
    /// conn.exec(|fs| {
    ///     fs.create("example.txt", FileKind::Regular, Owner::current())?;
    ///     liteboxfs::Result::Ok(())
    /// })?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn exec<T, E, F>(&mut self, f: F) -> Result<T, E>
    where
        F: FnOnce(&mut Filesystem) -> Result<T, E>,
        E: From<crate::Error>,
    {
        self.tx()?.exec(f)
    }

    /// Execute the given function within a new transaction with the given
    /// [`TransactionBehavior`].,
    ///
    /// See [`Transaction::exec`].
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Connection, CreateOptions, FileKind, Owner, TransactionBehavior};
    /// # let opts = CreateOptions::new();
    /// # let mut conn = Connection::open_in_memory(&opts)?;
    /// conn.exec_with(TransactionBehavior::Immediate, |fs| {
    ///     fs.create("example.txt", FileKind::Regular, Owner::current())?;
    ///     liteboxfs::Result::Ok(())
    /// })?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn exec_with<T, E, F>(&mut self, behavior: TransactionBehavior, f: F) -> Result<T, E>
    where
        F: FnOnce(&mut Filesystem) -> Result<T, E>,
        E: From<crate::Error>,
    {
        self.tx_with(behavior)?.exec(f)
    }

    /// Change the secret (password or key) used to encrypt the litebox via SQLCipher.
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Connection, CreateOptions, EncryptionSecret};
    /// # let opts = CreateOptions::new().encryption_secret(EncryptionSecret::password("old"));
    /// # let mut conn = Connection::open_in_memory(&opts)?;
    /// let new_secret = EncryptionSecret::password("correct horse battery staple");
    /// conn.change_secret(&new_secret)?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    #[cfg(feature = "_encryption")]
    pub fn change_secret(&mut self, secret: &EncryptionSecret) -> crate::Result<()> {
        match &secret.inner {
            EncryptionMode::Password(password) => {
                use secrecy::ExposeSecret;
                self.conn
                    .pragma_update(None, "rekey", password.expose_secret())?;
                Ok(())
            }
            EncryptionMode::Key(key) => {
                use secrecy::ExposeSecret;
                self.conn.pragma_update(
                    None,
                    "rekey",
                    Self::key_to_hex_pragma(key.expose_secret()),
                )?;
                Ok(())
            }
        }
    }

    /// Get the current SQLite journal mode.
    pub fn journal_mode(&mut self) -> crate::Result<JournalMode> {
        Ok(self
            .conn
            .pragma_query_value(None, "journal_mode", |row| row.get(0))?)
    }

    /// Set the SQLite journal mode.
    ///
    /// This can be used to enable [write-ahead logging](https://sqlite.org/wal.html).
    pub fn set_journal_mode(&mut self, mode: JournalMode) -> crate::Result<()> {
        self.conn.pragma_update(None, "journal_mode", mode)?;
        Ok(())
    }

    /// Mount `root` to `mountpoint` on the host filesystem using FUSE.
    ///
    /// You can specify a particular `root_path` in the root to mount. Pass `"/"` to mount the root
    /// directory.
    ///
    /// To enforce file permission checks, you must mount with [`MountOption::DefaultPermissions`].
    /// Note that when you first create a litebox, the root directory is owned by the root user
    /// with `755` permissions. This means if you mount the root directory with
    /// [`MountOption::DefaultPermissions`], the mounting user might not have write permissions for
    /// the root directory.
    ///
    /// Mounting a litebox acquires an exclusive lock, meaning that other liteboxfs connections
    /// cannot open it while it is mounted. This is **not** an exclusive SQLite transaction, which
    /// means that other processes **can** still open the underlying SQLite database, but they
    /// should never touch any tables used by LiteboxFS.
    ///
    /// This method blocks until the filesystem is unmounted.
    ///
    /// # Errors
    ///
    /// - [`RootNotFound`]: The given `root` does not exist.
    /// - [`NotADirectory`]: The given `root_path` is not a directory.
    /// - [`LiteboxLocked`]: An exclusive lock could not be acquired because the litebox is
    ///   currently open in another connection.
    ///
    /// [`RootNotFound`]: crate::Error::RootNotFound
    /// [`NotADirectory`]: crate::Error::NotADirectory
    /// [`LiteboxLocked`]: crate::Error::LiteboxLocked
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use liteboxfs::{Connection, CreateOptions, MountOption, RootBy};
    /// # let opts = CreateOptions::new();
    /// # let conn = Connection::create("example.litebox", &opts)?;
    /// conn.mount(RootBy::Default, "/", "/mnt/example", [MountOption::DefaultPermissions])?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    #[cfg(all(feature = "fuse", target_os = "linux"))]
    #[cfg_attr(coverage_nightly, coverage(off))]
    pub fn mount<'r, R, P, M, O>(
        mut self,
        root: R,
        root_path: P,
        mountpoint: M,
        options: O,
    ) -> crate::Result<()>
    where
        R: Into<RootBy<'r>>,
        P: AsRef<Path>,
        M: AsRef<Path>,
        O: IntoIterator<Item = MountOption>,
    {
        // We disable test coverage reporting for this method because:
        // 1. The fact that it blocks makes it difficult to test in via unit tests.
        // 2. We already test the FUSE implementation via external tests.

        use crate::fuse::FuseAdapter;
        use std::{mem, path::PathBuf};

        let mut options = options.into_iter().collect::<Vec<_>>();

        let has_allow_other = options
            .iter()
            .any(|option| option == &MountOption::AllowOther);

        let has_allow_root = options
            .iter()
            .any(|option| option == &MountOption::AllowRoot);

        let mut config = fuser::Config::default();

        options
            .retain(|option| !matches!(option, MountOption::AllowOther | MountOption::AllowRoot));

        config.mount_options = options.into_iter().map(MountOption::into_fuser).collect();

        config.acl = if has_allow_other {
            fuser::SessionACL::All
        } else if has_allow_root {
            fuser::SessionACL::RootAndOwner
        } else {
            fuser::SessionACL::Owner
        };

        let db_path = self.conn.path().map(PathBuf::from);

        let root_id = match root.into() {
            RootBy::Id(id) => id,
            RootBy::Default => self.default_root_id,
            RootBy::Name(name) => self.exec(|fs| crate::Result::Ok(fs.find_root(name)?.id))?,
        };

        // Upgrade the shared litebox lock held by this connection to an exclusive one, so that no
        // other connection can modify the litebox while the FUSE adapter is running. The exclusive
        // handle must outlive `fuser::mount2`, which blocks until unmount; binding it to a local
        // keeps it alive until this function returns.
        let _exclusive_lock = match mem::replace(&mut self.litebox_lock, LiteboxLock::Mounted) {
            LiteboxLock::OnDisk(handle) => match handle.try_upgrade()? {
                Ok(exclusive) => Some(exclusive),
                Err(_shared) => return Err(crate::Error::LiteboxLocked),
            },
            LiteboxLock::InMemory => None,
            LiteboxLock::Mounted => unreachable!(),
        };

        let adapter = FuseAdapter::new(self, root_id, root_path.as_ref(), db_path)?;

        fuser::mount2(adapter, mountpoint, &config)?;

        Ok(())
    }
}