liteboxfs 0.2.0

use std::{
    fmt::{self, Debug},
    io::{self, Read, Seek, Write},
    mem,
    ops::Range,
    path::Path,
    time::SystemTime,
};

#[cfg(all(feature = "fs", target_os = "linux"))]
use crate::RawFd;

use crate::{
    FileOrigin, RootId,
    block::{
        BlockData, BlockList, BlockSignature, BlockStoreInput, FileId as StoreFileId, FileOffset,
        LogicalBlockStoreAdapter, MerkleNodeStore,
    },
    cache::CachedBlock,
    file_metadata::{ACCESS_ACL_XATTR_NAME, DEFAULT_ACL_XATTR_NAME},
    id::{ContentId, FileId},
    lock::{LockHandle, LockHandler, LockResult, LockType},
    merkle::MerkleStore,
    metadata::{Acl, FileKind, FileMetadata, FileMode, Gid, Uid, Xattrs},
    path::NormalizedPath,
    settings::Settings,
    sql::{ExclusiveFileId, FileDiscriminant, PathInsertResult, SqlStore, SqlStoreGuard},
    state::{FileState, RegularFileStateContext, RegularFileStateGuard},
};

#[derive(Clone, Copy)]
enum FileIdState {
    Shared(StoreFileId),
    Exclusive(ExclusiveFileId),
}

impl FileIdState {
    fn store_id(self) -> StoreFileId {
        match self {
            Self::Shared(id) => id,
            Self::Exclusive(exclusive) => exclusive.file_id(),
        }
    }
}

// Flush data buffered in the chunker.
//
// After calling this, you MUST remember to store the returned block list back into the cache.
fn flush_chunker(
    state: &mut RegularFileStateContext<'_>,
    store: &mut SqlStore<'_>,
    exclusive: ExclusiveFileId,
) -> crate::Result<BlockList> {
    let mut block_list = match mem::take(state.cached_block_list) {
        Some(list) => list,
        None => store.data_store().list_blocks(exclusive.file_id())?,
    };

    while let Some(chunk) = state.chunker.next_chunk_or_remaining() {
        block_list =
            store.write_file_block(exclusive, *state.physical_pos, block_list, chunk.into())?;

        *state.physical_pos += chunk.len() as u64;
    }

    Ok(block_list)
}

/// A handle for accessing a file in the filesystem.
///
/// This type allows for reading and writing the contents and metadata of a file.
///
/// You can create a file with [`Filesystem::create`] or open an existing file with
/// [`Filesystem::open`]. You can open a file either by its path or by its [`FileId`].
///
/// # Reading and Writing
///
/// This type represents a regular file, directory, or special file. It implements [`Read`],
/// [`Write`], and [`Seek`] for reading and writing regular files, however these methods will
/// return a [`NotARegularFile`] error if called on a directory or special file.
///
/// # File Types
///
/// You can see whether this is a regular file, directory, or special file using [`File::kind`]. If
/// this file is a symlink, you can access its target through the returned [`FileKind`]. If this
/// file is a block or char device, you can access its device numbers through the [`FileKind`].
///
/// # Metadata
///
/// When accessing a litebox through this Rust API, reading and writing file contents and metadata
/// does not automatically update the [accessed](FileMetadata::accessed),
/// [modified](FileMetadata::modified), or [changed](FileMetadata::changed) times. Similarly, file
/// permissions and ACLs are not enforced by this Rust API. However, mounting a litebox through
/// FUSE *does* touch file times and enforce permissions.
///
/// File times in LiteboxFS have nanosecond precision.
///
/// [`Filesystem::create`]: crate::Filesystem::create
/// [`Filesystem::open`]: crate::Filesystem::open
/// [`Filesystem::delete`]: crate::Filesystem::delete
/// [`Filesystem::unlink`]: crate::Filesystem::unlink
/// [`NotARegularFile`]: crate::Error::NotARegularFile
/// [`Deferred`]: crate::TransactionBehavior::Deferred
pub struct File<'conn, 'fs> {
    root_id: RootId,
    file_id: FileIdState,
    kind: FileKind,
    settings: Settings,
    store: &'fs mut SqlStoreGuard<'conn>,
    state: FileState,
    lock: Option<LockHandle>,
    lock_handler: LockHandler,
}

#[cfg_attr(coverage_nightly, coverage(off))]
impl<'conn, 'fs> Debug for File<'conn, 'fs> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut debug = f.debug_struct("File");

        debug
            .field("root_id", &self.root_id)
            .field("file_id", &self.file_id.store_id());

        match &self.state {
            FileState::Regular(state) => debug
                .field("logical_pos", &state.logical_pos())
                .field("physical_pos", &state.physical_pos())
                .finish(),
            FileState::Special => debug.finish(),
        }
    }
}

pub(super) struct FileInput<'conn, 'fs> {
    pub root_id: RootId,
    pub file_id: StoreFileId,
    pub kind: FileKind,
    pub settings: Settings,
    pub store: &'fs mut SqlStoreGuard<'conn>,
    pub lock: LockHandle,
    pub lock_handler: LockHandler,
}

impl<'conn, 'fs> File<'conn, 'fs> {
    pub(super) fn new(input: FileInput<'conn, 'fs>) -> Self {
        let discriminant = input.kind.discriminant();

        Self {
            root_id: input.root_id,
            file_id: FileIdState::Shared(input.file_id),
            kind: input.kind,
            store: input.store,
            state: if matches!(discriminant, FileDiscriminant::Regular) {
                FileState::Regular(RegularFileStateGuard::new(&input.settings))
            } else {
                FileState::Special
            },
            settings: input.settings,
            lock: Some(input.lock),
            lock_handler: input.lock_handler,
        }
    }

    /// Ensure this file's metadata row is private to this root, performing a COW copy if needed.
    ///
    /// After this call, `self.file_id` is updated to the exclusive ID. Subsequent mutations reuse
    /// the cached `ExclusiveFileId` without a second materialization check.
    fn ensure_exclusive(&mut self) -> crate::Result<ExclusiveFileId> {
        match self.file_id {
            FileIdState::Exclusive(exclusive) => Ok(exclusive),
            FileIdState::Shared(file_id) => {
                let root_id = self.root_id;
                let exclusive = self
                    .store
                    .exec(|store| store.materialize_file(file_id, root_id))?;
                self.file_id = FileIdState::Exclusive(exclusive);
                Ok(exclusive)
            }
        }
    }

    /// Get a [`FileId`] which uniquely identifies the file in the filesystem.
    pub fn file_id(&self) -> FileId {
        FileId::new(self.file_id.store_id(), self.settings.uuid)
    }

    /// The [`FileKind`] of this file.
    pub fn kind(&self) -> &FileKind {
        // We can store this information in the file handle when we open it to avoid an extra
        // query. This is safe to do because file IDs are never recycled, so the file underlying
        // this file handle can never be changed out from under us.
        &self.kind
    }

    /// The apparent size of the file in bytes.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is not a regular file.
    ///
    /// # See Also
    ///
    /// - [`File::is_empty`]
    ///
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    ///
    /// # Examples
    ///
    /// ```
    /// # use std::io::Write;
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// file.write_all(b"Hello, world!")?;
    /// assert_eq!(file.len()?, 13);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn len(&mut self) -> crate::Result<u64> {
        let exclusive = self.ensure_exclusive()?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                });
            }
        };

        state.with_rewind(|mut state| {
            self.store.exec(|store| {
                let mut block_list = flush_chunker(&mut state, store, exclusive)?;

                let file_len = block_list.file_len();
                *state.cached_block_list = Some(block_list);

                Ok(file_len)
            })
        })
    }

    /// Return whether the apparent size of this file is zero.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is not a regular file.
    ///
    /// # See Also
    ///
    /// - [`File::len`]
    ///
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    pub fn is_empty(&mut self) -> crate::Result<bool> {
        Ok(self.len()? == 0)
    }

    /// Truncate or extend the object to the given `len`.
    ///
    /// If the given `len` is greater than the current length of the file, the object will be
    /// extended to `len` and the intermediate space will be filled with null bytes. This creates a
    /// sparse hole in the object, so no additional space is used in the database.
    ///
    /// If `len` is less than the current length of the file and the seek position is past the
    /// point which the file is truncated to, it is moved to the new end of the file.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is not a regular file.
    /// - [`TooLarge`]: The given length exceeds the maximum file size.
    ///
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    /// [`TooLarge`]: crate::Error::TooLarge
    ///
    /// # Examples
    ///
    /// Extend a file with a sparse hole:
    ///
    /// ```
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// file.set_len(1024)?;
    /// assert_eq!(file.len()?, 1024);
    /// # liteboxfs::Result::Ok(())
    /// ```
    ///
    /// Truncate a file:
    ///
    /// ```
    /// # use std::io::Write;
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// file.write_all(b"Hello, world!")?;
    /// file.set_len(5)?;
    /// assert_eq!(file.len()?, 5);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn set_len(&mut self, len: u64) -> crate::Result<()> {
        use std::cmp::Ordering;

        let exclusive = self.ensure_exclusive()?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                });
            }
        };

        state.with_rewind(|mut state| {
            self.store.exec(|store| {
                let mut block_list = flush_chunker(&mut state, store, exclusive)?;
                let current_len = block_list.file_len();

                *state.cached_block_list = Some(match len.cmp(&current_len) {
                    Ordering::Equal => return Ok(()),
                    Ordering::Less => {
                        state
                            .merkle_dirty_ranges
                            .mark_dirty_truncate(current_len, len);

                        let new_block_list = store.truncate_file(exclusive, block_list, len)?;

                        if *state.logical_pos > len {
                            *state.logical_pos = len;
                        }

                        if *state.physical_pos > len {
                            *state.physical_pos = len;
                        }

                        new_block_list
                    }
                    Ordering::Greater => store.write_file_block(
                        exclusive,
                        current_len,
                        block_list,
                        BlockStoreInput::Hole {
                            len: len - current_len,
                        },
                    )?,
                });

                Ok(())
            })
        })
    }

    /// The locations of sparse holes in the file.
    ///
    /// This returns the byte ranges of sparse holes, which are created via [`File::set_len`] or
    /// seeking past the end of the file.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is not a regular file.
    ///
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    ///
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// file.set_len(1024)?;
    /// assert_eq!(file.holes()?, vec![0..1024]);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn holes(&mut self) -> crate::Result<Vec<Range<u64>>> {
        let exclusive = self.ensure_exclusive()?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                });
            }
        };

        state.with_rewind(|mut state| {
            self.store.exec(|store| {
                let block_list = flush_chunker(&mut state, store, exclusive)?;
                let mut holes = Vec::<Range<u64>>::new();

                block_list.iter().fold(0, |offset, block| {
                    if let BlockSignature::Hole { len } = block.signature {
                        let hole_start = offset;
                        let hole_end = offset + len;

                        if let Some(last_hole) = holes.last_mut()
                            && last_hole.end == hole_start
                        {
                            // Merge adjacent holes.
                            last_hole.end = hole_end;
                        } else {
                            holes.push(hole_start..hole_end);
                        }
                    }

                    offset + block.len() as u64
                });

                *state.cached_block_list = Some(block_list);

                Ok(holes)
            })
        })
    }

    /// Get a [`ContentId`] which uniquely identifies the content of the file.
    ///
    /// Computing a [`ContentId`] can be expensive. If you need to determine if two files have
    /// identical contents, compare them by [`File::len`] first.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is not a regular file.
    ///
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    ///
    /// # Examples
    ///
    /// ```
    /// # use std::io::Write;
    /// # 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()?;
    /// let mut a = fs.create("a.txt", FileKind::Regular, Owner::current())?;
    /// a.write_all(b"Hello")?;
    /// let a_id = a.content_id()?;
    /// drop(a);
    /// let mut b = fs.create("b.txt", FileKind::Regular, Owner::current())?;
    /// b.write_all(b"Hello")?;
    /// assert_eq!(a_id, b.content_id()?);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn content_id(&mut self) -> crate::Result<ContentId> {
        let exclusive = self.ensure_exclusive()?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                });
            }
        };

        state.with_rewind(|mut state| {
            self.store.exec(|store| {
                // Update the database with the list of dirty logical blocks we've been tracking in
                // memory. This is necessary to ensure the merkle tree in the database reflects the
                // current state of the file.
                let dirty_ranges = state.merkle_dirty_ranges.take_ranges_collapsed();
                for dirty_range in dirty_ranges {
                    store.mark_dirty(exclusive.file_id(), dirty_range)?;
                }

                let mut block_list = flush_chunker(&mut state, store, exclusive)?;

                let adapter =
                    LogicalBlockStoreAdapter::new(store, self.settings.logical_block_size);
                let mut merkle_store = MerkleStore::new(adapter, &self.settings);

                let merkle_hash =
                    merkle_store.compute_hash(exclusive.file_id(), &mut block_list)?;

                *state.cached_block_list = Some(block_list);

                Ok(ContentId::new(merkle_hash, self.settings.uuid))
            })
        })
    }

    /// Return the number of hard links to this file.
    ///
    /// The number of hard links includes the file itself; you can think of this as the number of
    /// paths this file has in the filesystem. Unless the file has been unlinked via
    /// [`Filesystem::unlink`] or is a temporary file created with [`Filesystem::create_temp`],
    /// this count will be at least 1.
    ///
    /// # See Also
    ///
    /// - [`File::link`]
    ///
    /// [`Filesystem::unlink`]: crate::Filesystem::unlink
    /// [`Filesystem::create_temp`]: crate::Filesystem::create_temp
    ///
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// assert_eq!(file.link_count()?, 1);
    /// file.link("hardlink.txt")?;
    /// assert_eq!(file.link_count()?, 2);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn link_count(&mut self) -> crate::Result<u32> {
        self.store
            .exec(|store| store.count_paths(self.root_id, self.file_id.store_id()))
    }

    /// Create a hard link to this file at `link`.
    ///
    /// Hard links, unlike symbolic links, do not actually distinguish between the "link" and the
    /// "target". Creating a hard link just gives an existing file a second path in the filesystem.
    /// A file is deleted from the filesystem once it no longer has any hard links.
    ///
    /// You can determine if two files are hard links by comparing their [`FileId`]s.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is a directory. Hard links to directories are not
    ///   supported.
    /// - [`FileAlreadyExists`]: A file already exists at `link`.
    /// - [`NoParentDirectory`]: The parent directory of `link` does not exist.
    /// - [`NotADirectory`]: The parent of `link` is not a directory.
    ///
    /// # See Also
    ///
    /// - [`File::link_count`]
    ///
    /// [`FileId`]: crate::FileId
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    /// [`FileAlreadyExists`]: crate::Error::FileAlreadyExists
    /// [`NoParentDirectory`]: crate::Error::NoParentDirectory
    /// [`NotADirectory`]: crate::Error::NotADirectory
    ///
    /// # Examples
    ///
    /// ```
    /// # use std::io::{Write, Read};
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// let file_id = file.file_id();
    /// write!(file, "Hello, world!")?;
    ///
    /// file.link("hardlink.txt")?;
    /// drop(file);
    ///
    /// let mut link = fs.open("hardlink.txt")?;
    /// let mut actual = String::new();
    /// link.read_to_string(&mut actual)?;
    ///
    /// assert_eq!(link.file_id(), file_id);
    /// assert_eq!(&actual, "Hello, world!");
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn link<P: AsRef<Path>>(&mut self, link: P) -> crate::Result<()> {
        if self.kind == FileKind::Dir {
            return Err(crate::Error::NotARegularFile {
                file: FileOrigin::Litebox {
                    root: self.root_id,
                    locator: self.file_id().into(),
                },
            });
        }

        let link = NormalizedPath::new(link.as_ref());

        self.store.exec(|store| {
            match store.insert_path(self.root_id, &link, self.file_id.store_id())? {
                PathInsertResult::Inserted => Ok(()),
                PathInsertResult::AlreadyExists => Err(crate::Error::FileAlreadyExists {
                    path: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: link.to_path_buf(),
                    },
                }),
            }
        })
    }

    /// The metadata for the file.
    pub fn metadata(&mut self) -> crate::Result<FileMetadata> {
        self.store.exec(|store| {
            let raw = store.get_file_metadata_by_id(self.file_id.store_id())?;
            Ok(FileMetadata::from_raw(raw))
        })
    }

    /// Set the file mode.
    pub fn set_mode(&mut self, mode: FileMode) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_mode(exclusive, mode))
    }

    /// Set the owning user of the file.
    pub fn set_user(&mut self, uid: Uid) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_uid(exclusive, uid))
    }

    /// Set the owning group of the file.
    pub fn set_group(&mut self, gid: Gid) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_gid(exclusive, gid))
    }

    /// Set the time the file was last accessed.
    pub fn set_accessed(&mut self, atime: SystemTime) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_atime(exclusive, atime))
    }

    /// Set the time the file's contents were last changed.
    pub fn set_modified(&mut self, mtime: SystemTime) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_mtime(exclusive, mtime))
    }

    /// Set the time the file's metadata was last changed.
    pub fn set_changed(&mut self, ctime: SystemTime) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_ctime(exclusive, ctime))
    }

    /// Set the time the file was originally created.
    pub fn set_created(&mut self, btime: Option<SystemTime>) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.update_file_btime(exclusive, btime))
    }

    /// Update the file's modified, accessed, and changed times to `time`.
    ///
    /// # Examples
    ///
    /// ```
    /// # use std::time::{Duration, UNIX_EPOCH};
    /// # 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()?;
    /// let when = UNIX_EPOCH + Duration::from_secs(1_700_000_000);
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// file.touch_at(when)?;
    /// let metadata = file.metadata()?;
    /// assert_eq!(metadata.modified(), when);
    /// assert_eq!(metadata.accessed(), when);
    /// assert_eq!(metadata.changed(), when);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn touch_at(&mut self, time: SystemTime) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store.exec(|store| {
            store.update_file_mtime(exclusive, time)?;
            store.update_file_atime(exclusive, time)?;
            store.update_file_ctime(exclusive, time)?;

            crate::Result::Ok(())
        })
    }

    /// Update the file's modified, accessed, and changed times to now.
    ///
    /// # 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()?;
    /// let mut file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// let before = file.metadata()?.modified();
    /// file.touch()?;
    /// assert!(file.metadata()?.modified() >= before);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn touch(&mut self) -> crate::Result<()> {
        self.touch_at(SystemTime::now())
    }

    /// The extended attributes for the file.
    ///
    /// Note that because ACLs are stored as extended attributes, the returned [`Xattrs`] will
    /// include the xattrs that store the file's ACLs, which have the names
    /// [`ACCESS_ACL_XATTR_NAME`] and [`DEFAULT_ACL_XATTR_NAME`].
    ///
    /// # See Also
    ///
    /// - [`File::set_xattrs`]
    ///
    /// [`ACCESS_ACL_XATTR_NAME`]: crate::ACCESS_ACL_XATTR_NAME
    /// [`DEFAULT_ACL_XATTR_NAME`]: crate::DEFAULT_ACL_XATTR_NAME
    pub fn xattrs(&mut self) -> crate::Result<Xattrs> {
        self.store
            .exec(|store| store.get_file_xattrs(self.file_id.store_id()))
    }

    /// Set the extended attributes for the file.
    ///
    /// Note that because ACLs are stored as extended attributes, clearing a file's xattrs will
    /// also clear its ACLs. The xattrs that store the file's ACLs have the names
    /// [`ACCESS_ACL_XATTR_NAME`] and [`DEFAULT_ACL_XATTR_NAME`].
    ///
    /// # See Also
    ///
    /// - [`File::xattrs`]
    ///
    /// [`ACCESS_ACL_XATTR_NAME`]: crate::ACCESS_ACL_XATTR_NAME
    /// [`DEFAULT_ACL_XATTR_NAME`]: crate::DEFAULT_ACL_XATTR_NAME
    pub fn set_xattrs(&mut self, xattrs: &Xattrs) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        self.store
            .exec(|store| store.set_file_xattrs(exclusive, xattrs))
    }

    /// The access control list which defines the current access permissions.
    ///
    /// # See Also
    ///
    /// - [`File::set_access_acl`]
    /// - [`File::default_acl`]
    /// - [`File::set_default_acl`]
    pub fn access_acl(&mut self) -> crate::Result<Acl> {
        self.store.exec(|store| {
            match store.get_file_xattr(self.file_id.store_id(), ACCESS_ACL_XATTR_NAME.as_bytes())? {
                Some(data) => Acl::deserialize(&data),
                None => Ok(Acl::new()),
            }
        })
    }

    /// Set the access control list which defines the current access permissions.
    ///
    /// # See Also
    ///
    /// - [`File::access_acl`]
    /// - [`File::default_acl`]
    /// - [`File::set_default_acl`]
    pub fn set_access_acl(&mut self, acl: &Acl) -> crate::Result<()> {
        let exclusive = self.ensure_exclusive()?;
        let serialized = acl.serialize();
        self.store.exec(|store| {
            store.set_file_xattr(exclusive, ACCESS_ACL_XATTR_NAME.as_bytes(), &serialized)
        })
    }

    /// The access control list which defines the access permissions inherited by descendants.
    ///
    /// # See Also
    ///
    /// - [`File::access_acl`]
    /// - [`File::set_access_acl`]
    /// - [`File::set_default_acl`]
    pub fn default_acl(&mut self) -> crate::Result<Acl> {
        self.store.exec(|store| {
            match store
                .get_file_xattr(self.file_id.store_id(), DEFAULT_ACL_XATTR_NAME.as_bytes())?
            {
                Some(data) => Acl::deserialize(&data),
                None => Ok(Acl::new()),
            }
        })
    }

    /// Set the access control list which defines the access permissions inherited by descendants.
    ///
    /// Setting this value only makes sense for directories.
    ///
    /// # Errors
    ///
    /// - [`NotADirectory`]: This file is not a directory.
    ///
    /// # See Also
    ///
    /// - [`File::access_acl`]
    /// - [`File::set_access_acl`]
    /// - [`File::default_acl`]
    ///
    /// [`NotADirectory`]: crate::Error::NotADirectory
    ///
    /// # Examples
    ///
    /// ```
    /// # use liteboxfs::{Acl, AclMode, AclQualifier, 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()?;
    /// let mut dir = fs.create("example", FileKind::Dir, Owner::current())?;
    /// let mut acl = Acl::new();
    /// acl.set(AclQualifier::Other, AclMode::R);
    /// dir.set_default_acl(&acl)?;
    /// assert_eq!(dir.default_acl()?, acl);
    /// # liteboxfs::Result::Ok(())
    /// ```
    pub fn set_default_acl(&mut self, acl: &Acl) -> crate::Result<()> {
        // Default ACLs only make sense for directories.
        if !matches!(self.kind, FileKind::Dir) {
            return Err(crate::Error::NotADirectory {
                file: FileOrigin::Litebox {
                    root: self.root_id,
                    locator: self.file_id().into(),
                },
            });
        }

        let exclusive = self.ensure_exclusive()?;
        let serialized = acl.serialize();
        self.store.exec(|store| {
            store.set_file_xattr(exclusive, DEFAULT_ACL_XATTR_NAME.as_bytes(), &serialized)
        })
    }

    /// Consume this file and return a [`RawFd`].
    ///
    /// A [`RawFd`] can be used to keep this file from being deleted without holding an exclusive
    /// reference to the filesystem. See [`Filesystem::unlink`] and [`RawFd`] for more details.
    ///
    /// Because only regular files can be unlinked, you can only get a [`RawFd`] for a regular
    /// file.
    ///
    /// # Errors
    ///
    /// - [`NotARegularFile`]: This file is not a regular file.
    ///
    /// # See Also
    ///
    /// - [`Filesystem::unlink`]
    ///
    /// [`Filesystem::unlink`]: crate::Filesystem::unlink
    /// [`NotARegularFile`]: crate::Error::NotARegularFile
    ///
    /// # 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()?;
    /// let file = fs.create("example.txt", FileKind::Regular, Owner::current())?;
    /// let file_id = file.file_id();
    /// let fd = file.leak_fd()?;
    /// fs.unlink("example.txt")?;
    /// assert_eq!(fs.open(file_id)?.file_id(), file_id);
    /// fs.release(fd)?;
    /// # liteboxfs::Result::Ok(())
    /// ```
    #[cfg(all(feature = "fs", target_os = "linux"))]
    pub fn leak_fd(mut self) -> crate::Result<RawFd> {
        if !matches!(self.kind, FileKind::Regular) {
            return Err(crate::Error::NotARegularFile {
                file: FileOrigin::Litebox {
                    root: self.root_id,
                    locator: self.file_id().into(),
                },
            });
        }

        Ok(RawFd::new(
            self.lock
                .take()
                .expect("Expected this file to have a lock handle."),
            self.file_id.store_id(),
            self.settings.uuid,
        ))
    }
}

impl<'conn, 'fs> Read for File<'conn, 'fs> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let exclusive = self.ensure_exclusive().map_err(io::Error::from)?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                }
                .into());
            }
        };

        if buf.is_empty() {
            return Ok(0);
        }

        Ok(state.with_rewind(|mut state| {
            let bytes_filled = self.store.exec(|store| {
                // Update the database with the list of dirty logical blocks we've been tracking in
                // memory.
                let dirty_ranges = state.merkle_dirty_ranges.take_ranges_collapsed();
                for dirty_range in dirty_ranges {
                    store.mark_dirty(exclusive.file_id(), dirty_range)?;
                }

                let block_list = flush_chunker(&mut state, store, exclusive)?;

                let (block_offset, block) = match block_list.at_offset(*state.logical_pos) {
                    Some(block) => block,
                    None => {
                        return Ok(0);
                    }
                };

                *state.cached_block_list = Some(block_list);

                let mut data_store = store.data_store();

                match state.cached_block {
                    Some(cached_block) if cached_block.id() != block.id => {
                        cached_block.replace(block.id, |data| match data {
                            BlockData::Data { bytes } => {
                                bytes.clear();
                                let read_block = data_store.read_block(block.id, bytes)?;
                                if let BlockSignature::Hole { len } = read_block.signature {
                                    *data = BlockData::Hole { len }
                                }
                                crate::Result::Ok(())
                            }
                            BlockData::Hole { .. } => {
                                let mut buf =
                                    Vec::with_capacity(self.settings.largest_block_size());
                                let read_block = data_store.read_block(block.id, &mut buf)?;
                                match read_block.signature {
                                    BlockSignature::Data { .. } => {
                                        *data = BlockData::Data { bytes: buf }
                                    }
                                    BlockSignature::Hole { len } => *data = BlockData::Hole { len },
                                }
                                crate::Result::Ok(())
                            }
                        })?;
                    }
                    None => {
                        let mut buf = Vec::with_capacity(self.settings.largest_block_size());
                        let read_block = data_store.read_block(block.id, &mut buf)?;
                        *state.cached_block = Some(CachedBlock::new(
                            block.id,
                            match read_block.signature {
                                BlockSignature::Data { .. } => BlockData::Data { bytes: buf },
                                BlockSignature::Hole { len } => BlockData::Hole { len },
                            },
                        ));
                    }
                    _ => {}
                }

                if let Some(cached_block) = &mut state.cached_block {
                    Ok(cached_block.fill(buf, block_offset))
                } else {
                    unreachable!("A block should have been read and cached by now.");
                }
            })?;

            *state.logical_pos += bytes_filled as u64;
            *state.physical_pos = *state.logical_pos;

            crate::Result::Ok(bytes_filled)
        })?)
    }
}

impl<'conn, 'fs> Seek for File<'conn, 'fs> {
    fn seek(&mut self, pos: io::SeekFrom) -> io::Result<u64> {
        let exclusive = self.ensure_exclusive().map_err(io::Error::from)?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                }
                .into());
            }
        };

        Ok(state.with_rewind(|mut state| {
            self.store.exec(|store| {
                let mut block_list = flush_chunker(&mut state, store, exclusive)?;
                let current_len = block_list.file_len();

                *state.cached_block_list = Some(block_list);

                let new_pos = match pos {
                    io::SeekFrom::Start(offset) => crate::Result::Ok(offset),
                    io::SeekFrom::End(offset) => {
                        let new_pos = current_len.checked_add_signed(offset);

                        match new_pos {
                            Some(p) => Ok(p),
                            None => Err(io::Error::new(
                                io::ErrorKind::InvalidInput,
                                "Attempted to seek to a negative or overflowing position.",
                            )
                            .into()),
                        }
                    }
                    io::SeekFrom::Current(offset) => {
                        let new_pos = state.logical_pos.checked_add_signed(offset);

                        match new_pos {
                            Some(p) => Ok(p),
                            None => Err(io::Error::new(
                                io::ErrorKind::InvalidInput,
                                "Attempted to seek to a negative or overflowing position.",
                            )
                            .into()),
                        }
                    }
                }?;

                *state.logical_pos = new_pos;
                *state.physical_pos = *state.logical_pos;

                Ok(new_pos)
            })
        })?)
    }
}

impl<'conn, 'fs> Write for File<'conn, 'fs> {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        let exclusive = self.ensure_exclusive().map_err(io::Error::from)?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                }
                .into());
            }
        };

        if buf.is_empty() {
            return Ok(0);
        }

        Ok(state.with_rewind(|state| {
            state
                .merkle_dirty_ranges
                .mark_dirty_write(*state.logical_pos, buf.len() as FileOffset);

            let mut block_list = match mem::take(state.cached_block_list) {
                Some(list) => list,
                // It is possible to end up with two savepoints in a single write: one here, and
                // one when we actually write blocks to the database. This isn't ideal from a
                // performance perspective—savepoints are expensive—but it should be an uncommon
                // case. Once we've queried the block list, it should be cached for subsequent
                // writes.
                None => self
                    .store
                    .exec(|store| store.data_store().list_blocks(exclusive.file_id()))?,
            };

            let file_len = block_list.file_len();

            let written_pos = if *state.logical_pos < file_len {
                // If we're writing into the middle of the file, write the entire buffer and
                // let the data store handle re-chunking as needed.
                block_list = self.store.exec(|store| {
                    store.write_file_block(exclusive, *state.logical_pos, block_list, buf.into())
                })?;

                *state.logical_pos + buf.len() as u64
            } else {
                // If were appending to the file, we chunk the data ourselves. The reason why
                // we let the data store handle chunking when overwriting but handle it here
                // when appending is so that we can chunk across calls to `Write::write`,
                // maintaining the same chunker state.
                let mut running_pos = *state.physical_pos;

                state.chunker.push_bytes(buf);

                // We want to avoid starting a savepoint until we've saturated the chunker and it's
                // ready to produce more chunks. Savepoints are expensive, and we don't need to
                // start one if we're just writing to the chunker buffer.
                if state.chunker.buffer_size() >= self.settings.min_write_buffer_size {
                    block_list = self.store.exec(|store| {
                        while let Some(chunk) = state.chunker.next_chunk() {
                            block_list = store.write_file_block(
                                exclusive,
                                running_pos,
                                block_list,
                                chunk.into(),
                            )?;

                            running_pos += chunk.len() as u64;
                        }

                        Ok(block_list)
                    })?;
                }

                running_pos
            };

            *state.cached_block_list = Some(block_list);

            *state.logical_pos += buf.len() as u64;
            *state.physical_pos = written_pos;

            crate::Result::Ok(buf.len())
        })?)
    }

    fn flush(&mut self) -> io::Result<()> {
        let exclusive = self.ensure_exclusive().map_err(io::Error::from)?;

        let state = match &mut self.state {
            FileState::Regular(state) => state,
            FileState::Special => {
                return Err(crate::Error::NotARegularFile {
                    file: FileOrigin::Litebox {
                        root: self.root_id,
                        locator: self.file_id().into(),
                    },
                }
                .into());
            }
        };

        Ok(state.with_rewind(|mut state| {
            self.store.exec(|store| {
                let block_list = flush_chunker(&mut state, store, exclusive)?;

                *state.cached_block_list = Some(block_list);
                *state.physical_pos = *state.logical_pos;

                Ok(())
            })
        })?)
    }
}

impl<'conn, 'fs> Drop for File<'conn, 'fs> {
    fn drop(&mut self) {
        self.store
            .exec(|store| {
                // Only flush the chunker if writes occurred: if file_id is Exclusive, writes may
                // have been buffered in the chunker. If it is still Shared, no writes could have
                // happened so the chunker is empty and flushing is unnecessary.
                if let (FileState::Regular(state), FileIdState::Exclusive(exclusive)) =
                    (&mut self.state, self.file_id)
                {
                    state.with_rewind(|mut state| {
                        // Update the database with the list of dirty logical blocks we've been
                        // tracking in memory.
                        let dirty_ranges = state.merkle_dirty_ranges.take_ranges_collapsed();
                        for dirty_range in dirty_ranges {
                            store.mark_dirty(exclusive.file_id(), dirty_range)?;
                        }

                        // We don't need to worry about updating the cached block list here,
                        // because we're dropping it anyways.
                        flush_chunker(&mut state, store, exclusive)?;

                        crate::Result::Ok(())
                    })?;
                }

                // Release the read lock on this file so we can acquire a write lock. If another
                // file handle is being dropped at the same time and acquires a write lock after we
                // drop our read lock but before we acquire a write lock below, the effect is the
                // same: the file is deleted.
                if self.lock.take().is_none() {
                    // This must mean `File::leak_fd` has been called, in which case we should not
                    // attempt to delete the file.
                    return Ok(());
                }

                // Attempt to acquire an exclusive lock on the file to see if it's open in any
                // other transactions.
                match self
                    .lock_handler
                    .acquire_file_lock(self.file_id.store_id(), LockType::Write)?
                {
                    // No other transactions have this file open, so we can delete it if it's been
                    // unlinked. If the file has not been unlinked, this does nothing.
                    LockResult::Acquired(_lock_handle) => {
                        store.delete_file_if_unlinked(self.file_id.store_id())?;
                    }

                    // This file is open in at least one other transaction, so we cannot delete it
                    // yet.
                    LockResult::Locked => {}
                }

                Ok(())
            })
            .ok();
    }
}