//! This crate contains a Linux FUSE file-system implementation called LayerFS.
//!
//! # Introduction
//! The file-system is based on "layers" which can stack on top of each-other. Each layer
//! represents its own complete file-system, and when stacked the file-systems are unioned
//! together. When file-systems are unioned together the one which is stacked on "top" of the other
//! is favored. This achieves a similar effect to what can be done with overlay-fs.
//!
//! # Disk Layout
//! A layer is stored on disk as a directory of files. Each directory contains the following files
//! - `super.bin` contains information about the layer including any layers it is stacked on top of
//! - `file_table.bin` contains a listing of all the files in the layer
//! - `attributes_table.bin` contains the attributes for all the files in the layer
//! - `<offset>.dir_data.bin` contains directory contents for the directory found at `<offset>` in
//!    the file table.
//!
//! Notice that none of the above bullets mention containing file-data. Most file-data is instead
//! read from files outside of the layer. This what the `cache_dir` in [`LayerFs::from_path`] is
//! used for. However, if a file contains a small amount of data it can actually be stored in the
//! attributes as "inline data" to avoid the overhead of reading from another file.
//!
//! # The Stacking
//! When a layer is stacked on top of other layers, the directory entries for that layer may point
//! to directories or files in the lower layers. This is supported because [`FileId`] contains a
//! [`LayerId`]. This allows us to create the aforementioned overlay-fs like functionality.
//!
//! When an upper layer contains files which also exist in a lower layer, the lower layer files are
//! obscured by the upper layer files. When an upper layer contains a directory that also exists in
//! the lower layer, the directory contents are unioned together.
//!
//! ```ascii art
//! +------------------------+             +-----------+
//! | /   layer_id 2         |  built from | /         |
//! | `-- a.txt => (2, 10)   | <========== | `-- a.txt |
//! | `-- b.txt => (2, 3)    |             +-----------+
//! | `-- c.txt => (1, 3)    |
//! | `-- d/ => (0, 3)       |
//! +------------------------+             +-----------+
//! | /   layer_id 1         |  built from | /         |
//! | `-- a.txt => (1, 10)   | <========== | `-- a.txt |
//! | `-- c.txt => (1, 3)    |             | `-- c.txt |
//! | `-- d/ => (0, 3)       |             +-----------+
//! +------------------------+
//! | /   layer_id 0         |
//! | `-- d/ => (0, 3)       |
//! |     `-- e.txt => (0, 4)|
//! +------------------------+
//! ```
//! figure 1. Three stacked layers
//!
//! In figure 1 we illustrate a stacking of three layers. We show the [`FileId`] each directory
//! entry has as a way to illustrate how the stacking functions. Here is a description of the
//! layers
//! - Layer 0 has a directory `/d` and file `/d/e.txt`. Since this is a bottom layer all of its
//!   directory entries have `LayerId` 0
//! - Layer 1 has two files `/a.txt` and `/c.txt`. It also has a directory entry for `/d` from
//!   layer 0. Since this layer contains no entry itself for `/d`, this was merged in.
//! - Layer 2 has files `/a.txt`, `/b.txt`, `/c.txt` and directory `/d`. It has inherited entries
//!   for `/b.txt`, `/c.txt` and `/d` This can be seen because they all have `FileId`s which
//!   point to layer 1 or 0. `/a.txt` has `LayerId` 2, meaning that this file is shadowing `/a.txt`
//!   from layer 1. This means the `/a.txt` in layer 1 is now inaccessible.
//!
//! # Building Layers
//! Conceptually there are two different kinds of layers, "bottom layers" and "upper layers".
//! Upper layers are layer which are stacked on top of other layers. Bottom layers are layers which
//! aren't stacked on top of any other layers and are thus self-sufficient. Upper layers require
//! the layers they are stacked on top of to be present in order to be used. Upper layers are
//! created by stacking a bottom layer on top of some other layer.
//!
//! ```ascii art
//! +----------------+
//! | 2 upper layer  | <-------+
//! +----------------+         |
//! +----------------+         | UpperLayerBuilder::fill_from_bottom_layer
//! | 1 upper layer  |         |
//! +----------------+         |
//! +----------------+  +---------------+
//! | 0 bottom layer |  |  bottom layer |
//! +----------------+  +---------------+
//! ```
//! figure 2. A new upper layer with id 2 is being created stacked on top of two other layers.
//!
//! When creating a bottom layer, [`BottomLayerBuilder`] must be used. It can create a bottom layer
//! using either a manifest or a tar file as input.
//!
//! When creating an upper layer, [`UpperLayerBuilder`] must be used. The path to the layer you are
//! stacking on top of is given as input to [`UpperLayerBuilder::new`]. The bottom layer we are
//! using as input to create the upper layer is passed to
//! [`UpperLayerBuilder::fill_from_bottom_layer`].
//!
//! When creating upper layers, the layers that they are stacked on top of are not modified. This
//! means that layers can be reused in multiple stacks. This is also why the layer numbering is
//! done from the bottom-up.
//!
//! # Serving the File-System
//! Serving the file-system is done via FUSE. First the path to the top of the layer stack you want
//! to serve should be passed to [`LayerFs::from_path`]. Then either [`LayerFs::mount`] or
//! [`LayerFs::run_fuse`] should be called.

mod avl;
mod builder;
mod dir;
mod file;
mod ty;

use anyhow::{anyhow, Result};
use anyhow_trace::anyhow_trace;
pub use builder::*;
pub use dir::DirectoryDataReader;
pub use file::FileMetadataReader;
use futures::stream::StreamExt as _;
use lru::LruCache;
use maelstrom_base::Sha256Digest;
use maelstrom_fuse::{
    AttrResponse, EntryResponse, ErrnoResult, FileAttr, FuseFileSystem, ReadLinkResponse,
    ReadResponse, Request,
};
use maelstrom_linux::Errno;
use maelstrom_util::async_fs::Fs;
use std::ffi::OsStr;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{MappedMutexGuard, Mutex, MutexGuard};
pub use ty::{FileAttributes, FileData, FileId, FileType, LayerId, LayerSuper};

const TTL: Duration = Duration::from_secs(1); // 1 second
                                              //
fn to_eio<ValueT, ErrorT: std::fmt::Debug>(
    log: slog::Logger,
    res: std::result::Result<ValueT, ErrorT>,
) -> ErrnoResult<ValueT> {
    res.map_err(|err| {
        slog::error!(log, "Got error servicing FUSE request. Returning EIO"; "error" => ?err);
        Errno::EIO
    })
}

fn to_einval<ValueT, ErrorT: std::fmt::Debug>(
    log: slog::Logger,
    res: std::result::Result<ValueT, ErrorT>,
) -> ErrnoResult<ValueT> {
    res.map_err(|err| {
        slog::error!(log, "Got error servicing FUSE request. Returning EIO"; "error" => ?err);
        Errno::EINVAL
    })
}

enum LazyLayerSuperInner {
    NotCached(PathBuf),
    Cached(LayerSuper),
}

struct LazyLayerSuper(Mutex<LazyLayerSuperInner>);

impl LazyLayerSuper {
    fn not_cached(path: PathBuf) -> Self {
        Self(Mutex::new(LazyLayerSuperInner::NotCached(path)))
    }

    fn cached(layer_super: LayerSuper) -> Self {
        Self(Mutex::new(LazyLayerSuperInner::Cached(layer_super)))
    }

    async fn read(&self, data_fs: &Fs) -> Result<MappedMutexGuard<'_, LayerSuper>> {
        let mut inner = self.0.lock().await;
        if let LazyLayerSuperInner::NotCached(path) = &*inner {
            let new_state =
                LazyLayerSuperInner::Cached(LayerSuper::read_from_path(data_fs, path).await?);
            *inner = new_state;
        }
        Ok(MutexGuard::map(inner, |r| {
            let LazyLayerSuperInner::Cached(s) = r else {
                unreachable!()
            };
            s
        }))
    }
}

/// The in-memory representation of a mountable `LayerFs` layer. This can be either a bottom layer
/// or upper layer.
pub struct LayerFs {
    data_fs: Fs,
    top_layer_path: PathBuf,
    layer_super: LazyLayerSuper,
    cache_path: PathBuf,
}

#[anyhow_trace]
impl LayerFs {
    /// Instantiate a `LayerFs` with the given path to its data structures. `cache_path` should
    /// contain a path to a directory that can be used to look-up file-data via digest.
    pub fn from_path(data_dir: &Path, cache_path: &Path) -> Result<Self> {
        let data_fs = Fs::new();
        let data_dir = data_dir.to_owned();

        Ok(Self {
            data_fs,
            layer_super: LazyLayerSuper::not_cached(data_dir.join("super.bin")),
            top_layer_path: data_dir,
            cache_path: cache_path.to_owned(),
        })
    }

    async fn new(data_dir: &Path, cache_path: &Path, layer_super: LayerSuper) -> Result<Self> {
        let data_fs = Fs::new();
        let data_dir = data_dir.to_owned();

        layer_super
            .write_to_path(&data_fs, &data_dir.join("super.bin"))
            .await?;

        Ok(Self {
            data_fs,
            top_layer_path: data_dir,
            layer_super: LazyLayerSuper::cached(layer_super),
            cache_path: cache_path.to_owned(),
        })
    }

    async fn root(&self) -> Result<FileId> {
        Ok(FileId::root(self.layer_super().await?.layer_id))
    }

    async fn data_path(&self, layer_id: LayerId) -> Result<PathBuf> {
        if layer_id == self.layer_super().await?.layer_id {
            Ok(self.top_layer_path.clone())
        } else {
            self.layer_super()
                .await?
                .lower_layers
                .get(&layer_id)
                .ok_or(anyhow!("unknown layer {layer_id:?}"))
                .cloned()
        }
    }

    async fn dir_data_path(&self, mut file_id: FileId) -> Result<PathBuf> {
        if file_id.is_root() {
            file_id = self.root().await?;
        }
        Ok(self
            .data_path(file_id.layer())
            .await?
            .join(format!("{}.dir_data.bin", file_id.offset())))
    }

    async fn file_table_path(&self, layer_id: LayerId) -> Result<PathBuf> {
        Ok(self.data_path(layer_id).await?.join("file_table.bin"))
    }

    async fn attributes_table_path(&self, layer_id: LayerId) -> Result<PathBuf> {
        Ok(self.data_path(layer_id).await?.join("attributes_table.bin"))
    }

    async fn layer_super(&self) -> Result<LayerSuper> {
        Ok(self.layer_super.read(&self.data_fs).await?.clone())
    }

    pub async fn layer_id(&self) -> Result<LayerId> {
        Ok(self.layer_super().await?.layer_id)
    }

    fn cache_entry(&self, digest: &Sha256Digest) -> PathBuf {
        self.cache_path.join(digest.to_string())
    }

    /// Mount the file-system in a child process. It can then be accessed via a path in `/proc`.
    /// See [`maelstrom_fuse::fuse_mount_namespace`] for more details.
    pub async fn mount(
        self,
        log: slog::Logger,
        cache: Arc<Mutex<ReaderCache>>,
    ) -> Result<maelstrom_fuse::FuseNamespaceHandle> {
        slog::debug!(log, "mounting FUSE file-system in namespace");
        let adapter = LayerFsFuseAdapter::new(self, log.clone(), cache);
        maelstrom_fuse::fuse_mount_namespace(adapter, log, "Maelstrom LayerFS").await
    }

    /// Serve the file-system using the given FUSE file-descriptor. The function returns when the
    /// connection has closed either with an error or due to unmounting.
    pub async fn run_fuse(
        self,
        log: slog::Logger,
        cache: Arc<Mutex<ReaderCache>>,
        fd: maelstrom_linux::OwnedFd,
    ) -> Result<()> {
        slog::debug!(log, "running FUSE file-system");
        let adapter = LayerFsFuseAdapter::new(self, log.clone(), cache);
        maelstrom_fuse::run_fuse(adapter, log, fd).await
    }
}

/// This cache contains a synchronized collection of open files and cached data.
pub struct ReaderCache {
    dir_readers: LruCache<PathBuf, Arc<Mutex<DirectoryDataReader>>>,
    file_readers: LruCache<PathBuf, Arc<Mutex<FileMetadataReader>>>,
    data_files: LruCache<Sha256Digest, Arc<std::fs::File>>,
}

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

impl ReaderCache {
    pub fn new() -> Self {
        Self {
            dir_readers: LruCache::new(300.try_into().unwrap()),
            file_readers: LruCache::new(300.try_into().unwrap()),
            data_files: LruCache::new(200.try_into().unwrap()),
        }
    }

    async fn open_dir(
        &mut self,
        layer_fs: &LayerFs,
        file_id: FileId,
    ) -> Result<Arc<Mutex<DirectoryDataReader>>> {
        let path = layer_fs.dir_data_path(file_id).await?;
        if let Some(reader) = self.dir_readers.get(&path) {
            Ok(reader.clone())
        } else {
            let reader = Arc::new(Mutex::new(
                DirectoryDataReader::new(layer_fs, file_id).await?,
            ));
            self.dir_readers.push(path, reader.clone());
            Ok(reader)
        }
    }

    async fn files(
        &mut self,
        layer_fs: &LayerFs,
        layer_id: LayerId,
    ) -> Result<Arc<Mutex<FileMetadataReader>>> {
        let path = layer_fs.file_table_path(layer_id).await?;
        if let Some(reader) = self.file_readers.get(&path) {
            Ok(reader.clone())
        } else {
            let reader = Arc::new(Mutex::new(
                FileMetadataReader::new(layer_fs, layer_id).await?,
            ));
            self.file_readers.push(path, reader.clone());
            Ok(reader)
        }
    }

    async fn data_file(
        &mut self,
        layer_fs: &LayerFs,
        digest: &Sha256Digest,
    ) -> Result<Arc<std::fs::File>> {
        if let Some(file) = self.data_files.get(digest) {
            Ok(file.clone())
        } else {
            let file = layer_fs
                .data_fs
                .open_file(layer_fs.cache_entry(digest))
                .await?;
            let file = Arc::new(file.into_std().await);
            self.data_files.push(digest.clone(), file.clone());
            Ok(file)
        }
    }
}

struct LayerFsFuseAdapter {
    layer_fs: LayerFs,
    log: slog::Logger,
    cache: Arc<Mutex<ReaderCache>>,
}

impl LayerFsFuseAdapter {
    fn new(layer_fs: LayerFs, log: slog::Logger, cache: Arc<Mutex<ReaderCache>>) -> Self {
        Self {
            layer_fs,
            log,
            cache,
        }
    }
}

impl FuseFileSystem for LayerFsFuseAdapter {
    async fn look_up(&self, req: Request, parent: u64, name: &OsStr) -> ErrnoResult<EntryResponse> {
        let name = to_einval(self.log.clone(), name.to_str().ok_or("invalid name"))?;
        let parent = to_einval(self.log.clone(), FileId::try_from(parent))?;
        let reader = to_eio(
            self.log.clone(),
            self.cache
                .lock()
                .await
                .open_dir(&self.layer_fs, parent)
                .await,
        )?;
        let child_id = to_eio(self.log.clone(), reader.lock().await.look_up(name).await)?
            .ok_or(Errno::ENOENT)?;
        let attrs = self.get_attr(req, child_id.as_u64()).await?;
        Ok(EntryResponse {
            attr: attrs.attr,
            ttl: TTL,
            generation: 0,
        })
    }

    async fn get_attr(&self, _req: Request, ino: u64) -> ErrnoResult<AttrResponse> {
        let file = to_einval(self.log.clone(), FileId::try_from(ino))?;
        let reader = to_eio(
            self.log.clone(),
            self.cache
                .lock()
                .await
                .files(&self.layer_fs, file.layer())
                .await,
        )?;
        let (kind, attrs) = to_eio(self.log.clone(), reader.lock().await.get_attr(file).await)?;
        Ok(AttrResponse {
            ttl: TTL,
            attr: FileAttr {
                ino,
                size: attrs.size,
                blocks: 0,
                atime: attrs.mtime.into(),
                mtime: attrs.mtime.into(),
                ctime: attrs.mtime.into(),
                kind,
                perm: u32::from(attrs.mode) as u16,
                nlink: 1,
                uid: 1000,
                gid: 1000,
                rdev: 0,
                blksize: 512,
            },
        })
    }

    async fn read(
        &self,
        _req: Request,
        ino: u64,
        _fh: u64,
        offset: i64,
        size: u32,
        _flags: i32,
        _lock: Option<u64>,
    ) -> ErrnoResult<ReadResponse> {
        let file = to_einval(self.log.clone(), FileId::try_from(ino))?;
        let reader = to_eio(
            self.log.clone(),
            self.cache
                .lock()
                .await
                .files(&self.layer_fs, file.layer())
                .await,
        )?;
        let (kind, data) = to_eio(self.log.clone(), reader.lock().await.get_data(file).await)?;
        if kind != FileType::RegularFile {
            return Err(Errno::EINVAL);
        }
        match data {
            FileData::Empty => Ok(ReadResponse::Buffer { data: vec![] }),
            FileData::Inline(inline) => {
                let offset = to_einval(self.log.clone(), usize::try_from(offset))?;
                if offset >= inline.len() {
                    return Err(Errno::EINVAL);
                }
                let size = std::cmp::min(size as usize, inline.len() - offset);

                Ok(ReadResponse::Buffer {
                    data: inline[offset..(offset + size)].to_vec(),
                })
            }
            FileData::Digest {
                digest,
                offset: file_start,
                length: file_length,
            } => {
                let read_start =
                    file_start + to_einval::<u64, _>(self.log.clone(), offset.try_into())?;
                let file_end = file_start + file_length;
                if read_start > file_end {
                    return Err(Errno::EINVAL);
                }

                let read_end = std::cmp::min(read_start + size as u64, file_end);
                let read_length = read_end - read_start;

                let file = to_eio(
                    self.log.clone(),
                    self.cache
                        .lock()
                        .await
                        .data_file(&self.layer_fs, &digest)
                        .await,
                )?;
                Ok(ReadResponse::Splice {
                    file,
                    offset: read_start,
                    length: read_length as usize,
                })
            }
        }
    }

    type ReadDirStream<'a> = Pin<
        Box<
            dyn futures::Stream<Item = maelstrom_fuse::ErrnoResult<maelstrom_fuse::DirEntry>>
                + Send,
        >,
    >;

    async fn read_dir(
        &self,
        _req: Request,
        ino: u64,
        _fh: u64,
        offset: i64,
    ) -> ErrnoResult<Self::ReadDirStream<'_>> {
        let file = to_einval(self.log.clone(), FileId::try_from(ino))?;
        let reader = to_eio(
            self.log.clone(),
            DirectoryDataReader::new(&self.layer_fs, file).await,
        )?;
        let stream = to_eio(
            self.log.clone(),
            reader.into_stream(offset.try_into()?).await,
        )?;
        let log = self.log.clone();
        Ok(Box::pin(stream.map(move |res| {
            let (offset, entry) = to_eio(log.clone(), res)?;
            Ok(maelstrom_fuse::DirEntry {
                ino: entry.value.file_id.as_u64(),
                offset: i64::try_from(offset).unwrap(),
                kind: entry.value.kind,
                name: entry.key,
            })
        })))
    }

    async fn read_link(&self, _req: Request, ino: u64) -> ErrnoResult<ReadLinkResponse> {
        let file = to_einval(self.log.clone(), FileId::try_from(ino))?;
        let reader = to_eio(
            self.log.clone(),
            self.cache
                .lock()
                .await
                .files(&self.layer_fs, file.layer())
                .await,
        )?;
        let (kind, data) = to_eio(self.log.clone(), reader.lock().await.get_data(file).await)?;
        if kind != FileType::Symlink {
            return Err(Errno::EINVAL);
        }
        match data {
            FileData::Empty => Ok(ReadLinkResponse { data: vec![] }),
            FileData::Inline(inline) => Ok(ReadLinkResponse {
                data: inline.to_vec(),
            }),
            FileData::Digest { .. } => Err(Errno::EIO),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use maelstrom_base::{
        manifest::{ManifestEntry, ManifestEntryData, ManifestEntryMetadata, Mode},
        Utf8PathBuf,
    };
    use maelstrom_util::manifest::AsyncManifestWriter;
    use slog::Drain as _;
    use std::future::Future;
    use std::os::unix::{ffi::OsStrExt as _, fs::MetadataExt as _};
    use std::pin::Pin;
    use tokio::io::AsyncWriteExt as _;

    const ARBITRARY_TIME: maelstrom_base::manifest::UnixTimestamp =
        maelstrom_base::manifest::UnixTimestamp(1705000271);

    struct BuildEntry {
        path: String,
        type_: FileType,
        data: FileData,
        mode: u32,
        is_hard_link: bool,
    }

    impl BuildEntry {
        fn reg(path: impl Into<String>, data: impl Into<Vec<u8>>) -> Self {
            Self::reg_mode(path, FileData::Inline(data.into()), 0o555)
        }

        fn reg_digest(
            path: impl Into<String>,
            digest: Sha256Digest,
            offset: u64,
            length: u64,
        ) -> Self {
            Self::reg_mode(
                path,
                FileData::Digest {
                    digest,
                    offset,
                    length,
                },
                0o555,
            )
        }

        fn reg_empty(path: impl Into<String>) -> Self {
            Self::reg_empty_mode(path, 0o555)
        }

        fn reg_empty_mode(path: impl Into<String>, mode: u32) -> Self {
            Self {
                path: path.into(),
                type_: FileType::RegularFile,
                data: FileData::Empty,
                mode,
                is_hard_link: false,
            }
        }

        fn reg_mode(path: impl Into<String>, data: FileData, mode: u32) -> Self {
            Self {
                path: path.into(),
                type_: FileType::RegularFile,
                data,
                mode,
                is_hard_link: false,
            }
        }

        fn dir(path: impl Into<String>) -> Self {
            Self::dir_mode(path, 0o555)
        }

        fn dir_mode(path: impl Into<String>, mode: u32) -> Self {
            Self {
                path: path.into(),
                type_: FileType::Directory,
                data: FileData::Empty,
                mode,
                is_hard_link: false,
            }
        }

        fn sym(path: impl Into<String>, target: impl Into<Vec<u8>>) -> Self {
            Self {
                path: path.into(),
                type_: FileType::Symlink,
                data: FileData::Inline(target.into()),
                mode: 0o777,
                is_hard_link: false,
            }
        }

        fn link(path: impl Into<String>, target: impl Into<Vec<u8>>) -> Self {
            Self {
                path: path.into(),
                type_: FileType::RegularFile,
                data: FileData::Inline(target.into()),
                mode: 0o777,
                is_hard_link: true,
            }
        }

        fn from_str(s: &str) -> Self {
            if s.ends_with("/") {
                Self::dir(s)
            } else {
                Self::reg_empty(s)
            }
        }
    }

    fn test_logger() -> slog::Logger {
        let decorator = slog_term::PlainSyncDecorator::new(slog_term::TestStdoutWriter);
        let drain = slog_term::FullFormat::new(decorator).build().fuse();
        let drain = slog_async::Async::new(drain).build().fuse();
        slog::Logger::root(drain, slog::o!())
    }

    async fn build_fs(
        fs: &Fs,
        data_dir: &Path,
        cache_path: &Path,
        files: Vec<BuildEntry>,
    ) -> LayerFs {
        let mut builder =
            BottomLayerBuilder::new(test_logger(), fs, data_dir, cache_path, ARBITRARY_TIME)
                .await
                .unwrap();

        for BuildEntry {
            path,
            type_,
            data,
            mode,
            is_hard_link,
        } in files
        {
            let size = match &data {
                ty::FileData::Empty => 0,
                ty::FileData::Inline(d) => d.len() as u64,
                ty::FileData::Digest { length, .. } => *length,
            };
            if is_hard_link {
                let FileData::Inline(data) = data else {
                    unreachable!()
                };

                builder
                    .add_link_path(path.as_ref(), std::str::from_utf8(&data).unwrap().as_ref())
                    .await
                    .unwrap();
                continue;
            }

            match type_ {
                FileType::RegularFile => {
                    builder
                        .add_file_path(
                            path.as_ref(),
                            ty::FileAttributes {
                                size,
                                mode: Mode(mode),
                                mtime: ARBITRARY_TIME,
                            },
                            data,
                        )
                        .await
                        .unwrap();
                }
                FileType::Directory => {
                    builder
                        .add_dir_path(
                            path.as_ref(),
                            ty::FileAttributes {
                                size,
                                mode: Mode(mode),
                                mtime: ARBITRARY_TIME,
                            },
                        )
                        .await
                        .unwrap();
                }
                FileType::Symlink => {
                    let FileData::Inline(data) = data else {
                        unreachable!()
                    };
                    builder.add_symlink_path(path.as_ref(), data).await.unwrap();
                }
                other => panic!("unsupported file type {other:?}"),
            }
        }

        builder.finish().await.unwrap()
    }

    async fn assert_entries(fs: &Fs, path: &Path, expected: Vec<&str>) {
        let mut entry_stream = fs.read_dir(path).await.unwrap();
        let mut entries = vec![];
        while let Some(e) = entry_stream.next().await {
            let e = e.unwrap();
            let mut name = e.file_name();
            if e.file_type().await.unwrap().is_dir() {
                name.push("/");
            }
            entries.push(name);
        }
        entries.sort();
        assert_eq!(
            entries,
            Vec::from_iter(expected.into_iter().map(|e| std::ffi::OsString::from(e)))
        );
    }

    #[tokio::test]
    async fn read_dir_and_look_up() {
        let temp = tempfile::tempdir().unwrap();
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg_empty("/Foo"),
                BuildEntry::reg_empty("/Bar"),
                BuildEntry::reg_empty("/Baz"),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        assert_entries(&fs, &mount_path, vec!["Bar", "Baz", "Foo"]).await;

        fs.metadata(mount_path.join("Bar")).await.unwrap();
        fs.metadata(mount_path.join("Baz")).await.unwrap();
        fs.metadata(mount_path.join("Foo")).await.unwrap();

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn read_dir_multi_level() {
        let temp = tempfile::tempdir().unwrap();
        let mount_path = temp.path().join("mount");
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&mount_path).await.unwrap();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg_empty("/Foo/Bar/Baz"),
                BuildEntry::reg_empty("/Foo/Bin"),
                BuildEntry::reg_empty("/Foo/Bar/Qux"),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        assert_entries(&fs, &mount_path, vec!["Foo/"]).await;
        assert_entries(&fs, &mount_path.join("Foo"), vec!["Bar/", "Bin"]).await;
        assert_entries(&fs, &mount_path.join("Foo/Bar"), vec!["Baz", "Qux"]).await;

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn get_attr() {
        let temp = tempfile::tempdir().unwrap();
        let mount_path = temp.path().join("mount");
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&mount_path).await.unwrap();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg_empty("/Foo"),
                BuildEntry::reg_empty("/Bar"),
                BuildEntry::reg_empty("/Baz"),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        for name in ["Foo", "Bar", "Baz"] {
            let attrs = fs.metadata(mount_path.join(name)).await.unwrap();
            assert_eq!(attrs.len(), 0);
            assert_eq!(Mode(attrs.mode()), Mode(0o100555));
            assert_eq!(attrs.mtime(), ARBITRARY_TIME.into());
        }

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn hard_link() {
        let temp = tempfile::tempdir().unwrap();
        let mount_path = temp.path().join("mount");
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&mount_path).await.unwrap();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg_empty("/Foo"),
                BuildEntry::link("/Bar", "/Foo"),
                BuildEntry::link("/Baz", "/Bar"),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        let foo_attrs = fs.metadata(mount_path.join("Foo")).await.unwrap();

        for name in ["Foo", "Bar", "Baz"] {
            let attrs = fs.metadata(mount_path.join(name)).await.unwrap();
            assert_eq!(attrs.len(), 0);
            assert_eq!(Mode(attrs.mode()), Mode(0o100555));
            assert_eq!(attrs.mtime(), ARBITRARY_TIME.into());
            assert_eq!(attrs.ino(), foo_attrs.ino());
        }

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn read_inline() {
        let temp = tempfile::tempdir().unwrap();
        let mount_path = temp.path().join("mount");
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&mount_path).await.unwrap();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg("/Foo", b"hello world"),
                BuildEntry::reg_empty("/Bar"),
                BuildEntry::reg_empty("/Baz"),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        let contents = fs.read_to_string(mount_path.join("Foo")).await.unwrap();
        assert_eq!(contents, "hello world");

        let contents = fs.read_to_string(mount_path.join("Bar")).await.unwrap();
        assert_eq!(contents, "");

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn read_link() {
        let temp = tempfile::tempdir().unwrap();
        let mount_path = temp.path().join("mount");
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&mount_path).await.unwrap();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg("/Foo", b"hello world"),
                BuildEntry::sym("/Bar", "./Foo"),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        let contents = fs.read_to_string(mount_path.join("Foo")).await.unwrap();
        assert_eq!(contents, "hello world");

        assert!(fs
            .symlink_metadata(mount_path.join("Bar"))
            .await
            .unwrap()
            .is_symlink());
        let contents = fs.read_to_string(mount_path.join("Bar")).await.unwrap();
        assert_eq!(contents, "hello world");

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn read_digest() {
        let temp = tempfile::tempdir().unwrap();
        let mount_path = temp.path().join("mount");
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&mount_path).await.unwrap();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let temp_path = cache_dir.join("temp");
        let mut f = fs.create_file(&temp_path).await.unwrap();
        f.write_all(b"hello world").await.unwrap();
        drop(f);
        let digest = calc_digest(&fs, &temp_path).await;
        fs.rename(temp_path, cache_dir.join(digest.to_string()))
            .await
            .unwrap();

        let layer_fs = build_fs(
            &fs,
            &data_dir,
            &cache_dir,
            vec![
                BuildEntry::reg_digest("/Foo", digest.clone(), 0, 5),
                BuildEntry::reg_digest("/Bar", digest.clone(), 6, 5),
            ],
        )
        .await;

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(test_logger(), cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        let contents = fs.read_to_string(mount_path.join("Foo")).await.unwrap();
        assert_eq!(contents, "hello");

        let contents = fs.read_to_string(mount_path.join("Bar")).await.unwrap();
        assert_eq!(contents, "world");

        mount_handle.umount_and_join().await.unwrap();
    }

    async fn calc_digest(fs: &Fs, path: &Path) -> Sha256Digest {
        let mut f = fs.open_file(path).await.unwrap();
        let mut hasher = maelstrom_util::io::Sha256Stream::new(tokio::io::sink());
        tokio::io::copy(&mut f, &mut hasher).await.unwrap();
        hasher.finalize().1
    }

    async fn build_tar(fs: &Fs, cache_path: &Path, files: Vec<BuildEntry>) -> Sha256Digest {
        let tar_path = cache_path.join("temp.tar");
        let f = fs.create_file(&tar_path).await.unwrap();
        let mut ar = tokio_tar::Builder::new(f.into_inner());
        for BuildEntry {
            path,
            type_,
            data,
            mode,
            is_hard_link,
        } in files
        {
            let mut header = tokio_tar::Header::new_gnu();
            let data = match data {
                ty::FileData::Empty => vec![],
                ty::FileData::Inline(d) => d,
                _ => panic!(),
            };
            header.set_entry_type(match (is_hard_link, type_) {
                (true, _) => tokio_tar::EntryType::Link,
                (_, FileType::RegularFile) => tokio_tar::EntryType::Regular,
                (_, FileType::Directory) => tokio_tar::EntryType::Directory,
                (_, FileType::Symlink) => tokio_tar::EntryType::Symlink,
                (_, other) => panic!("unsupported entry type {other:?}"),
            });
            if type_ == FileType::Symlink || is_hard_link {
                header.set_size(0);
                header
                    .set_link_name(std::ffi::OsStr::from_bytes(&data))
                    .unwrap();
                ar.append_data(&mut header, path, tokio::io::empty())
                    .await
                    .unwrap();
            } else {
                header.set_size(data.len() as u64);
                header.set_mode(mode);
                ar.append_data(&mut header, path, &data[..]).await.unwrap();
            }
        }
        ar.finish().await.unwrap();

        let digest = calc_digest(fs, &tar_path).await;

        fs.rename(tar_path, cache_path.join(digest.to_string()))
            .await
            .unwrap();

        digest
    }

    async fn build_manifest(fs: &Fs, cache_path: &Path, files: Vec<BuildEntry>) -> PathBuf {
        let manifest_path = cache_path.join("temp.manifest");
        let f = fs.create_file(&manifest_path).await.unwrap();
        let mut builder = AsyncManifestWriter::new(f).await.unwrap();
        for BuildEntry {
            path,
            type_,
            data,
            mode,
            is_hard_link,
        } in files
        {
            let path: Utf8PathBuf = path.into();
            let size = match &data {
                ty::FileData::Empty => 0,
                ty::FileData::Inline(d) => d.len() as u64,
                ty::FileData::Digest { length, .. } => *length,
            };
            let metadata = ManifestEntryMetadata {
                size,
                mode: Mode(mode),
                mtime: ARBITRARY_TIME,
            };
            if is_hard_link {
                let ty::FileData::Inline(data) = data else {
                    unimplemented!()
                };
                builder
                    .write_entry(&ManifestEntry {
                        path,
                        metadata,
                        data: ManifestEntryData::Hardlink(
                            std::str::from_utf8(&data).unwrap().into(),
                        ),
                    })
                    .await
                    .unwrap();
                continue;
            }
            match type_ {
                FileType::Directory => builder
                    .write_entry(&ManifestEntry {
                        path,
                        metadata,
                        data: ManifestEntryData::Directory,
                    })
                    .await
                    .unwrap(),
                FileType::RegularFile => {
                    let data = match data {
                        ty::FileData::Empty => None,
                        ty::FileData::Inline(d) => {
                            let temp_path = cache_path.join("temp.bin");
                            fs.write(&temp_path, d).await.unwrap();
                            let digest = calc_digest(fs, &temp_path).await;
                            fs.rename(temp_path, cache_path.join(digest.to_string()))
                                .await
                                .unwrap();
                            Some(digest)
                        }
                        ty::FileData::Digest { digest, offset, .. } => {
                            assert_eq!(offset, 0);
                            Some(digest)
                        }
                    };
                    builder
                        .write_entry(&ManifestEntry {
                            path,
                            metadata,
                            data: ManifestEntryData::File(data),
                        })
                        .await
                        .unwrap();
                }
                FileType::Symlink => {
                    let ty::FileData::Inline(data) = data else {
                        unimplemented!()
                    };
                    builder
                        .write_entry(&ManifestEntry {
                            path,
                            metadata,
                            data: ManifestEntryData::Symlink(data),
                        })
                        .await
                        .unwrap();
                }
                other => panic!("unsupported entry type {other:?}"),
            }
        }
        drop(builder);

        let digest = calc_digest(fs, &manifest_path).await;

        let final_path = cache_path.join(digest.to_string());
        fs.rename(manifest_path, &final_path).await.unwrap();

        final_path
    }

    #[cfg(test)]
    async fn layer_from_tar_or_manifest(
        populate_fn: impl for<'a> FnOnce(
            &'a Fs,
            &'a Path,
            Vec<BuildEntry>,
            &'a mut BottomLayerBuilder,
        ) -> Pin<Box<dyn Future<Output = ()> + 'a>>,
    ) {
        let temp = tempfile::tempdir().unwrap();
        let data_dir = temp.path().join("data");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&data_dir).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let input = vec![
            BuildEntry::reg("Foo", b"hello world"),
            BuildEntry::dir("Qux"),
            BuildEntry::reg_empty("Bar/Baz"),
            BuildEntry::reg_empty("Bar/Bin"),
            BuildEntry::reg_empty("Qux/Fred"),
            BuildEntry::dir_mode("Bar", 0o666),
            BuildEntry::sym("Waldo", b"Foo"),
            BuildEntry::link("Thud", "/Foo"),
        ];

        let log = test_logger();
        let mut builder =
            BottomLayerBuilder::new(log.clone(), &fs, &data_dir, &cache_dir, ARBITRARY_TIME)
                .await
                .unwrap();

        populate_fn(&fs, &cache_dir, input, &mut builder).await;

        let layer_fs = builder.finish().await.unwrap();

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(log, cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        let contents = fs.read_to_string(mount_path.join("Foo")).await.unwrap();
        assert_eq!(contents, "hello world");

        let contents = fs.read_to_string(mount_path.join("Thud")).await.unwrap();
        assert_eq!(contents, "hello world");

        assert!(fs
            .symlink_metadata(mount_path.join("Waldo"))
            .await
            .unwrap()
            .is_symlink());
        let contents = fs.read_to_string(mount_path.join("Waldo")).await.unwrap();
        assert_eq!(contents, "hello world");

        let contents = fs.read_to_string(mount_path.join("Bar/Baz")).await.unwrap();
        assert_eq!(contents, "");

        assert_entries(
            &fs,
            &mount_path,
            vec!["Bar/", "Foo", "Qux/", "Thud", "Waldo"],
        )
        .await;
        assert_entries(&fs, &mount_path.join("Bar"), vec!["Baz", "Bin"]).await;
        assert_entries(&fs, &mount_path.join("Qux"), vec!["Fred"]).await;

        for p in [
            "Foo", "Qux", "Bar/Baz", "Bar/Bin", "Qux/Fred", "Waldo", "Thud",
        ] {
            let mode = fs.metadata(mount_path.join(p)).await.unwrap().mode();
            assert_eq!(Mode(mode & 0o777), Mode(0o555));
        }

        let mode = fs.metadata(mount_path.join("Bar")).await.unwrap().mode();
        assert_eq!(Mode(mode & 0o777), Mode(0o666));

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn layer_from_tar() {
        layer_from_tar_or_manifest(|fs, cache_dir, input, builder| {
            Box::pin(async move {
                let tar_digest = build_tar(&fs, &cache_dir, input).await;
                let tar_path = cache_dir.join(tar_digest.to_string());

                builder
                    .add_from_tar(tar_digest, fs.open_file(tar_path).await.unwrap())
                    .await
                    .unwrap();
            })
        })
        .await
    }

    #[tokio::test]
    async fn layer_from_manifest() {
        layer_from_tar_or_manifest(|fs, cache_dir, input, builder| {
            Box::pin(async move {
                let manifest_path = build_manifest(&fs, &cache_dir, input).await;
                builder
                    .add_from_manifest(fs.open_file(manifest_path).await.unwrap())
                    .await
                    .unwrap();
            })
        })
        .await
    }

    async fn two_layer_test(lower: Vec<&str>, upper: Vec<&str>, expected: Vec<(&str, Vec<&str>)>) {
        let temp = tempfile::tempdir().unwrap();
        let data_dir1 = temp.path().join("data1");
        let data_dir2 = temp.path().join("data2");
        let data_dir3 = temp.path().join("data3");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&data_dir1).await.unwrap();
        fs.create_dir(&data_dir2).await.unwrap();
        fs.create_dir(&data_dir3).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs1 = build_fs(
            &fs,
            &data_dir1,
            &cache_dir,
            lower.into_iter().map(BuildEntry::from_str).collect(),
        )
        .await;

        let layer_fs2 = build_fs(
            &fs,
            &data_dir2,
            &cache_dir,
            upper.into_iter().map(BuildEntry::from_str).collect(),
        )
        .await;

        let log = test_logger();
        let mut builder = UpperLayerBuilder::new(log.clone(), &data_dir3, &cache_dir, &layer_fs1)
            .await
            .unwrap();
        builder.fill_from_bottom_layer(&layer_fs2).await.unwrap();
        let layer_fs = builder.finish().await.unwrap();

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(log, cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        for (d, entries) in expected {
            assert_entries(&fs, &mount_path.join(d), entries).await;
        }

        mount_handle.umount_and_join().await.unwrap();
    }

    async fn three_layer_test(
        lowest: Vec<&str>,
        lower: Vec<&str>,
        upper: Vec<&str>,
        expected: Vec<(&str, Vec<&str>)>,
    ) {
        let temp = tempfile::tempdir().unwrap();
        let data_dir1 = temp.path().join("data1");
        let data_dir2 = temp.path().join("data2");
        let data_dir3 = temp.path().join("data3");
        let data_dir4 = temp.path().join("data4");
        let data_dir5 = temp.path().join("data5");
        let cache_dir = temp.path().join("cache");

        let fs = Fs::new();
        fs.create_dir(&data_dir1).await.unwrap();
        fs.create_dir(&data_dir2).await.unwrap();
        fs.create_dir(&data_dir3).await.unwrap();
        fs.create_dir(&data_dir4).await.unwrap();
        fs.create_dir(&data_dir5).await.unwrap();
        fs.create_dir(&cache_dir).await.unwrap();

        let layer_fs1 = build_fs(
            &fs,
            &data_dir1,
            &cache_dir,
            lowest.into_iter().map(BuildEntry::from_str).collect(),
        )
        .await;

        let layer_fs2 = build_fs(
            &fs,
            &data_dir2,
            &cache_dir,
            lower.into_iter().map(BuildEntry::from_str).collect(),
        )
        .await;

        let layer_fs3 = build_fs(
            &fs,
            &data_dir3,
            &cache_dir,
            upper.into_iter().map(BuildEntry::from_str).collect(),
        )
        .await;

        let log = test_logger();

        let mut builder = UpperLayerBuilder::new(log.clone(), &data_dir4, &cache_dir, &layer_fs1)
            .await
            .unwrap();
        builder.fill_from_bottom_layer(&layer_fs2).await.unwrap();
        let upper_layer_fs = builder.finish().await.unwrap();

        let mut builder =
            UpperLayerBuilder::new(log.clone(), &data_dir5, &cache_dir, &upper_layer_fs)
                .await
                .unwrap();
        builder.fill_from_bottom_layer(&layer_fs3).await.unwrap();
        let layer_fs = builder.finish().await.unwrap();

        let cache = Arc::new(Mutex::new(ReaderCache::new()));
        let mount_handle = layer_fs.mount(log, cache).await.unwrap();
        let mount_path = mount_handle.mount_path();

        for (d, entries) in expected {
            assert_entries(&fs, &mount_path.join(d), entries).await;
        }

        mount_handle.umount_and_join().await.unwrap();
    }

    #[tokio::test]
    async fn two_layer_empty_bottom() {
        two_layer_test(
            vec![],
            vec!["/Pie/KeyLime", "/Cake/Birthday", "/Cookies/Snickerdoodle"],
            vec![
                ("", vec!["Cake/", "Cookies/", "Pie/"]),
                ("Cake", vec!["Birthday"]),
                ("Pie", vec!["KeyLime"]),
                ("Cookies", vec!["Snickerdoodle"]),
            ],
        )
        .await;
    }

    #[tokio::test]
    async fn two_layer_empty_top() {
        two_layer_test(
            vec!["/Pie/KeyLime", "/Cake/Birthday", "/Cookies/Snickerdoodle"],
            vec![],
            vec![
                ("", vec!["Cake/", "Cookies/", "Pie/"]),
                ("Cake", vec!["Birthday"]),
                ("Pie", vec!["KeyLime"]),
                ("Cookies", vec!["Snickerdoodle"]),
            ],
        )
        .await;
    }

    #[tokio::test]
    async fn two_layer_large_bottom_dir() {
        two_layer_test(
            vec!["/a/b/c/d/e/f"],
            vec!["/a/g"],
            vec![("a", vec!["b/", "g"]), ("a/b/c/d/e", vec!["f"])],
        )
        .await;
    }

    #[tokio::test]
    async fn two_layer_large_upper_dir() {
        two_layer_test(
            vec!["/a/g"],
            vec!["/a/b/c/d/e/f"],
            vec![("a", vec!["b/", "g"]), ("a/b/c/d/e", vec!["f"])],
        )
        .await;
    }

    #[tokio::test]
    async fn two_layer_complicated() {
        two_layer_test(
            vec!["/Pie/Apple", "/Cake/Chocolate", "/Cake/Cupcakes/Sprinkle"],
            vec!["/Pie/KeyLime", "/Cake/Birthday", "/Cookies/Snickerdoodle"],
            vec![
                ("", vec!["Cake/", "Cookies/", "Pie/"]),
                ("Cake", vec!["Birthday", "Chocolate", "Cupcakes/"]),
                ("Cake/Cupcakes", vec!["Sprinkle"]),
                ("Pie", vec!["Apple", "KeyLime"]),
                ("Cookies", vec!["Snickerdoodle"]),
            ],
        )
        .await;
    }

    #[tokio::test]
    async fn two_layer_replace_dir_with_file() {
        two_layer_test(
            vec!["/Cake/Cupcakes/Sprinkle"],
            vec!["/Cake/Cupcakes"],
            vec![("", vec!["Cake/"]), ("Cake", vec!["Cupcakes"])],
        )
        .await;
    }

    #[tokio::test]
    async fn two_layer_replace_file_with_dir() {
        two_layer_test(
            vec!["/Cake/Cupcakes"],
            vec!["/Cake/Cupcakes/Sprinkle"],
            vec![
                ("", vec!["Cake/"]),
                ("Cake", vec!["Cupcakes/"]),
                ("Cake/Cupcakes", vec!["Sprinkle"]),
            ],
        )
        .await;
    }

    #[tokio::test]
    async fn three_layer() {
        three_layer_test(
            vec!["/Cake/Cupcakes/"],
            vec!["/Cake/Chocolate/"],
            vec!["/Cake/Chocolate/"],
            vec![
                ("", vec!["Cake/"]),
                ("Cake", vec!["Chocolate/", "Cupcakes/"]),
            ],
        )
        .await;
    }
}
