common/bucket/

mount.rs

use std::collections::BTreeMap;
use std::io::Read;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use parking_lot::Mutex;
use uuid::Uuid;

use crate::crypto::{PublicKey, Secret, SecretError, SecretKey, Share};
use crate::linked_data::{BlockEncoded, CodecError, Link};
use crate::peer::{BlobsStore, BlobsStoreError};

use super::manifest::Manifest;
use super::node::{Node, NodeError, NodeLink};
use super::pins::Pins;

pub fn clean_path(path: &Path) -> PathBuf {
    if !path.is_absolute() {
        panic!("path is not absolute");
    }
    path.iter()
        .skip(1)
        .map(|part| part.to_string_lossy().to_string())
        .collect::<PathBuf>()
}

#[derive(Clone)]
pub struct MountInner {
    // link to the manifest
    pub link: Link,
    // the loaded manifest
    pub manifest: Manifest,
    // the loaded, decrypted entry node
    pub entry: Node,
    // the loaded pins
    pub pins: Pins,
}

impl MountInner {
    pub fn link(&self) -> &Link {
        &self.link
    }
    pub fn entry(&self) -> &Node {
        &self.entry
    }
    pub fn manifest(&self) -> &Manifest {
        &self.manifest
    }
    pub fn pins(&self) -> &Pins {
        &self.pins
    }
}

#[derive(Clone)]
pub struct Mount(Arc<Mutex<MountInner>>, BlobsStore);

#[derive(Debug, thiserror::Error)]
pub enum MountError {
    #[error("default error: {0}")]
    Default(#[from] anyhow::Error),
    #[error("link not found")]
    LinkNotFound(Link),
    #[error("path not found: {0}")]
    PathNotFound(PathBuf),
    #[error("path is not a node: {0}")]
    PathNotNode(PathBuf),
    #[error("blobs store error: {0}")]
    BlobsStore(#[from] BlobsStoreError),
    #[error("secret error: {0}")]
    Secret(#[from] SecretError),
    #[error("node error: {0}")]
    Node(#[from] NodeError),
    #[error("codec error: {0}")]
    Codec(#[from] CodecError),
    #[error("share error: {0}")]
    Share(#[from] crate::crypto::ShareError),
    #[error("peers share was not found. this should be impossible")]
    ShareNotFound,
}

impl Mount {
    pub fn inner(&self) -> MountInner {
        self.0.lock().clone()
    }

    pub fn blobs(&self) -> BlobsStore {
        self.1.clone()
    }

    pub fn link(&self) -> Link {
        let inner = self.0.lock();
        inner.link.clone()
    }

    /// Save the current mount state to the blobs store
    #[allow(clippy::await_holding_lock)]
    pub async fn save(&self, blobs: &BlobsStore) -> Result<Link, MountError> {
        let mut inner = self.0.lock();
        // Create a new secret for the updated root
        let secret = Secret::generate();
        // get the now previous link to the bucket
        let previous = inner.link.clone();
        // Put the current root node into blobs with the new secret
        let entry = Self::_put_node_in_blobs(&inner.entry, &secret, blobs).await?;
        // Serialize current pins to blobs
        // put the new root link into the pins, as well as the previous link
        inner.pins.insert(*entry.clone().hash());
        inner.pins.insert(*previous.hash());
        let pins_link = Self::_put_pins_in_blobs(&inner.pins, blobs).await?;
        // Update the bucket's share with the new root link
        // (add_share creates the Share internally)
        let mut manifest = inner.manifest.clone();
        let _m = manifest.clone();
        let shares = _m.shares();
        manifest.unset_shares();
        for share in shares.values() {
            let public_key = share.principal().identity;
            manifest.add_share(public_key, secret.clone())?;
        }
        // Update the bucket's pins field
        manifest.set_pins(pins_link.clone());
        manifest.set_previous(previous);
        manifest.set_entry(entry.clone());
        // Put the updated manifest into blobs to determine the new link
        let link = Self::_put_manifest_in_blobs(&manifest, blobs).await?;

        // update the internal state
        inner.manifest = manifest;

        Ok(link)
    }

    pub async fn init(
        id: Uuid,
        name: String,
        owner: &SecretKey,
        blobs: &BlobsStore,
    ) -> Result<Self, MountError> {
        // create a new root node for the bucket
        let entry = Node::default();
        // create a new secret for the owner
        let secret = Secret::generate();
        // put the node in the blobs store for the secret
        let entry_link = Self::_put_node_in_blobs(&entry, &secret, blobs).await?;
        // share the secret with the owner
        let share = Share::new(&secret, &owner.public())?;
        // Initialize pins with root node hash
        let mut pins = Pins::new();
        pins.insert(*entry_link.hash());
        // Put the pins in blobs to get a pins link
        let pins_link = Self::_put_pins_in_blobs(&pins, blobs).await?;
        // construct the new manifest
        let manifest = Manifest::new(
            id,
            name.clone(),
            owner.public(),
            share,
            entry_link.clone(),
            pins_link.clone(),
        );
        let link = Self::_put_manifest_in_blobs(&manifest, blobs).await?;

        // return the new mount
        Ok(Mount(
            Arc::new(Mutex::new(MountInner {
                link,
                manifest,
                entry,
                pins,
            })),
            blobs.clone(),
        ))
    }

    pub async fn load(
        link: &Link,
        secret_key: &SecretKey,
        blobs: &BlobsStore,
    ) -> Result<Self, MountError> {
        let public_key = &secret_key.public();
        let manifest = Self::_get_manifest_from_blobs(link, blobs).await?;

        let _share = manifest.get_share(public_key);

        let share = match _share {
            Some(share) => share.share(),
            None => return Err(MountError::ShareNotFound),
        };

        let secret = share.recover(secret_key)?;

        let pins = Self::_get_pins_from_blobs(manifest.pins(), blobs).await?;
        let entry =
            Self::_get_node_from_blobs(&NodeLink::Dir(manifest.entry().clone(), secret), blobs)
                .await?;

        Ok(Mount(
            Arc::new(Mutex::new(MountInner {
                link: link.clone(),
                manifest,
                entry,
                pins,
            })),
            blobs.clone(),
        ))
    }

    #[allow(clippy::await_holding_lock)]
    pub async fn share(&mut self, peer: PublicKey) -> Result<(), MountError> {
        let mut inner = self.0.lock();
        inner.manifest.add_share(peer, Secret::default())?;
        Ok(())
    }

    #[allow(clippy::await_holding_lock)]
    pub async fn add<R>(
        &mut self,
        path: &Path,
        data: R,
        blobs: &BlobsStore,
    ) -> Result<(), MountError>
    where
        R: Read + Send + Sync + 'static + Unpin,
    {
        let secret = Secret::generate();

        let encrypted_reader = secret.encrypt_reader(data)?;

        // TODO (amiller68): this is incredibly dumb
        use bytes::Bytes;
        use futures::stream;
        let encrypted_bytes = {
            let mut buf = Vec::new();
            let mut reader = encrypted_reader;
            reader.read_to_end(&mut buf).map_err(SecretError::Io)?;
            buf
        };

        let stream = Box::pin(stream::once(async move {
            Ok::<_, std::io::Error>(Bytes::from(encrypted_bytes))
        }));

        let hash = blobs.put_stream(stream).await?;

        let link = Link::new(
            crate::linked_data::LD_RAW_CODEC,
            hash,
            iroh_blobs::BlobFormat::Raw,
        );

        let node_link = NodeLink::new_data_from_path(link.clone(), secret, path);

        let mut inner = self.0.lock();
        let root_node = inner.entry.clone();
        let (updated_link, node_hashes) =
            Self::_set_node_link_at_path(root_node, node_link, path, blobs).await?;

        // Track pins: data blob + all created node hashes
        inner.pins.insert(hash);
        inner.pins.extend(node_hashes);

        if let NodeLink::Dir(new_root_link, new_secret) = updated_link {
            inner.entry = Self::_get_node_from_blobs(
                &NodeLink::Dir(new_root_link.clone(), new_secret),
                blobs,
            )
            .await?;
        }

        Ok(())
    }

    #[allow(clippy::await_holding_lock)]
    pub async fn rm(&mut self, path: &Path, blobs: &BlobsStore) -> Result<(), MountError> {
        let path = clean_path(path);
        let parent_path = path
            .parent()
            .ok_or_else(|| MountError::Default(anyhow::anyhow!("Cannot remove root")))?;

        let inner = self.0.lock();
        let entry = inner.entry.clone();
        drop(inner);

        let mut parent_node = if parent_path == Path::new("") {
            entry.clone()
        } else {
            Self::_get_node_at_path(&entry, parent_path, blobs).await?
        };

        let file_name = path.file_name().unwrap().to_string_lossy().to_string();

        if parent_node.del(&file_name).is_none() {
            return Err(MountError::PathNotFound(path.to_path_buf()));
        }

        if parent_path == Path::new("") {
            let secret = Secret::generate();
            let link = Self::_put_node_in_blobs(&parent_node, &secret, blobs).await?;

            let mut inner = self.0.lock();
            // Track the new root node hash
            inner.pins.insert(*link.hash());
            inner.entry = parent_node;
        } else {
            // Save the modified parent node to blobs
            let secret = Secret::generate();
            let parent_link = Self::_put_node_in_blobs(&parent_node, &secret, blobs).await?;
            let node_link = NodeLink::new_dir(parent_link.clone(), secret);

            let mut inner = self.0.lock();
            // Convert parent_path back to absolute for _set_node_link_at_path
            let abs_parent_path = Path::new("/").join(parent_path);
            let (updated_link, node_hashes) =
                Self::_set_node_link_at_path(entry, node_link, &abs_parent_path, blobs).await?;

            // Track the parent node hash and all created node hashes
            inner.pins.insert(*parent_link.hash());
            inner.pins.extend(node_hashes);

            if let NodeLink::Dir(new_root_link, new_secret) = updated_link {
                inner.entry = Self::_get_node_from_blobs(
                    &NodeLink::Dir(new_root_link.clone(), new_secret),
                    blobs,
                )
                .await?;
            }
        }

        Ok(())
    }

    #[allow(clippy::await_holding_lock)]
    pub async fn ls(
        &self,
        path: &Path,
        blobs: &BlobsStore,
    ) -> Result<BTreeMap<PathBuf, NodeLink>, MountError> {
        let mut items = BTreeMap::new();
        let path = clean_path(path);

        let inner = self.0.lock();
        let root_node = inner.entry.clone();
        drop(inner);

        let node = if path == Path::new("") {
            root_node
        } else {
            match Self::_get_node_at_path(&root_node, &path, blobs).await {
                Ok(node) => node,
                Err(MountError::LinkNotFound(_)) => {
                    return Err(MountError::PathNotNode(path.to_path_buf()))
                }
                Err(err) => return Err(err),
            }
        };

        for (name, link) in node.get_links() {
            let mut full_path = path.clone();
            full_path.push(name);
            items.insert(full_path, link.clone());
        }

        Ok(items)
    }

    pub async fn ls_deep(
        &self,
        path: &Path,
        blobs: &BlobsStore,
    ) -> Result<BTreeMap<PathBuf, NodeLink>, MountError> {
        let base_path = clean_path(path);
        self._ls_deep(path, &base_path, blobs).await
    }

    async fn _ls_deep(
        &self,
        path: &Path,
        base_path: &Path,
        blobs: &BlobsStore,
    ) -> Result<BTreeMap<PathBuf, NodeLink>, MountError> {
        let mut all_items = BTreeMap::new();

        // get the initial items at the given path
        let items = self.ls(path, blobs).await?;

        for (item_path, link) in items {
            // Make path relative to the base_path
            let relative_path = if base_path == Path::new("") {
                item_path.clone()
            } else {
                item_path
                    .strip_prefix(base_path)
                    .unwrap_or(&item_path)
                    .to_path_buf()
            };
            all_items.insert(relative_path.clone(), link.clone());

            if link.is_dir() {
                // Recurse using the absolute path
                let abs_item_path = Path::new("/").join(&item_path);
                let sub_items = Box::pin(self._ls_deep(&abs_item_path, base_path, blobs)).await?;

                // Sub items already have correct relative paths from base_path
                for (sub_path, sub_link) in sub_items {
                    all_items.insert(sub_path, sub_link);
                }
            }
        }

        Ok(all_items)
    }

    #[allow(clippy::await_holding_lock)]
    pub async fn cat(&self, path: &Path, blobs: &BlobsStore) -> Result<Vec<u8>, MountError> {
        let path = clean_path(path);

        let inner = self.0.lock();
        let root_node = inner.entry.clone();
        drop(inner);

        let (parent_path, file_name) = if let Some(parent) = path.parent() {
            (
                parent,
                path.file_name().unwrap().to_string_lossy().to_string(),
            )
        } else {
            return Err(MountError::PathNotFound(path.to_path_buf()));
        };

        let parent_node = if parent_path == Path::new("") {
            root_node
        } else {
            Self::_get_node_at_path(&root_node, parent_path, blobs).await?
        };

        let link = parent_node
            .get_link(&file_name)
            .ok_or_else(|| MountError::PathNotFound(path.to_path_buf()))?;

        match link {
            NodeLink::Data(link, secret, _) => {
                let encrypted_data = blobs.get(link.hash()).await?;
                let data = secret.decrypt(&encrypted_data)?;
                Ok(data)
            }
            NodeLink::Dir(_, _) => Err(MountError::PathNotNode(path.to_path_buf())),
        }
    }

    /// Get the NodeLink for a file at a given path
    #[allow(clippy::await_holding_lock)]
    pub async fn get(&self, path: &Path, blobs: &BlobsStore) -> Result<NodeLink, MountError> {
        let path = clean_path(path);

        let inner = self.0.lock();
        let root_node = inner.entry.clone();
        drop(inner);

        let (parent_path, file_name) = if let Some(parent) = path.parent() {
            (
                parent,
                path.file_name().unwrap().to_string_lossy().to_string(),
            )
        } else {
            return Err(MountError::PathNotFound(path.to_path_buf()));
        };

        let parent_node = if parent_path == Path::new("") {
            root_node
        } else {
            Self::_get_node_at_path(&root_node, parent_path, blobs).await?
        };

        parent_node
            .get_link(&file_name)
            .cloned()
            .ok_or_else(|| MountError::PathNotFound(path.to_path_buf()))
    }

    async fn _get_node_at_path(
        node: &Node,
        path: &Path,
        blobs: &BlobsStore,
    ) -> Result<Node, MountError> {
        let mut current_node = node.clone();
        let mut consumed_path = PathBuf::from("/");

        for part in path.iter() {
            consumed_path.push(part);
            let next = part.to_string_lossy().to_string();
            let next_link = current_node
                .get_link(&next)
                .ok_or(MountError::PathNotFound(consumed_path.clone()))?;
            current_node = Self::_get_node_from_blobs(next_link, blobs).await?
        }
        Ok(current_node)
    }

    pub async fn _set_node_link_at_path(
        node: Node,
        node_link: NodeLink,
        path: &Path,
        blobs: &BlobsStore,
    ) -> Result<(NodeLink, Vec<crate::linked_data::Hash>), MountError> {
        let path = clean_path(path);
        let mut visited_nodes = Vec::new();
        let mut name = path.file_name().unwrap().to_string_lossy().to_string();
        let parent_path = path.parent().unwrap_or(Path::new(""));

        let mut consumed_path = PathBuf::from("/");
        let mut node = node;
        visited_nodes.push((consumed_path.clone(), node.clone()));

        for part in parent_path.iter() {
            let next = part.to_string_lossy().to_string();
            let next_link = node.get_link(&next);
            if let Some(next_link) = next_link {
                consumed_path.push(part);
                match next_link {
                    NodeLink::Dir(..) => {
                        node = Self::_get_node_from_blobs(next_link, blobs).await?
                    }
                    NodeLink::Data(..) => {
                        return Err(MountError::PathNotNode(consumed_path.clone()));
                    }
                }
                visited_nodes.push((consumed_path.clone(), node.clone()));
            } else {
                // Create a new directory node
                node = Node::default();
                consumed_path.push(part);
                visited_nodes.push((consumed_path.clone(), node.clone()));
            }
        }

        let mut node_link = node_link;
        let mut created_hashes = Vec::new();
        for (path, mut node) in visited_nodes.into_iter().rev() {
            node.insert(name, node_link.clone());
            let secret = Secret::generate();
            let link = Self::_put_node_in_blobs(&node, &secret, blobs).await?;
            created_hashes.push(*link.hash());
            node_link = NodeLink::Dir(link, secret);
            name = path
                .file_name()
                .unwrap_or_default()
                .to_string_lossy()
                .to_string();
        }

        Ok((node_link, created_hashes))
    }

    async fn _get_manifest_from_blobs(
        link: &Link,
        blobs: &BlobsStore,
    ) -> Result<Manifest, MountError> {
        tracing::debug!(
            "_get_bucket_from_blobs: Checking for bucket data at link {:?}",
            link
        );
        let hash = link.hash();
        tracing::debug!("_get_bucket_from_blobs: Bucket hash: {}", hash);

        match blobs.stat(hash).await {
            Ok(true) => {
                tracing::debug!(
                    "_get_bucket_from_blobs: Bucket hash {} exists in blobs",
                    hash
                );
            }
            Ok(false) => {
                tracing::error!("_get_bucket_from_blobs: Bucket hash {} NOT FOUND in blobs - LinkNotFound error!", hash);
                return Err(MountError::LinkNotFound(link.clone()));
            }
            Err(e) => {
                tracing::error!(
                    "_get_bucket_from_blobs: Error checking bucket hash {}: {}",
                    hash,
                    e
                );
                return Err(e.into());
            }
        }

        tracing::debug!("_get_bucket_from_blobs: Reading bucket data from blobs");
        let data = blobs.get(hash).await?;
        tracing::debug!(
            "_get_bucket_from_blobs: Got {} bytes of bucket data",
            data.len()
        );

        let bucket_data = Manifest::decode(&data)?;
        tracing::debug!(
            "_get_bucket_from_blobs: Successfully decoded BucketData for bucket '{}'",
            bucket_data.name()
        );

        Ok(bucket_data)
    }

    async fn _get_pins_from_blobs(link: &Link, blobs: &BlobsStore) -> Result<Pins, MountError> {
        tracing::debug!("_get_pins_from_blobs: Checking for pins at link {:?}", link);
        let hash = link.hash();
        tracing::debug!("_get_pins_from_blobs: Pins hash: {}", hash);

        match blobs.stat(hash).await {
            Ok(true) => {
                tracing::debug!("_get_pins_from_blobs: Pins hash {} exists in blobs", hash);
            }
            Ok(false) => {
                tracing::error!(
                    "_get_pins_from_blobs: Pins hash {} NOT FOUND in blobs - LinkNotFound error!",
                    hash
                );
                return Err(MountError::LinkNotFound(link.clone()));
            }
            Err(e) => {
                tracing::error!(
                    "_get_pins_from_blobs: Error checking pins hash {}: {}",
                    hash,
                    e
                );
                return Err(e.into());
            }
        }

        tracing::debug!("_get_pins_from_blobs: Reading hash list from blobs");
        // Read hashes from the hash list blob
        let hashes = blobs.read_hash_list(*hash).await?;
        tracing::debug!(
            "_get_pins_from_blobs: Successfully read {} hashes from pinset",
            hashes.len()
        );

        Ok(Pins::from_vec(hashes))
    }

    async fn _get_node_from_blobs(
        node_link: &NodeLink,
        blobs: &BlobsStore,
    ) -> Result<Node, MountError> {
        let link = node_link.link();
        let secret = node_link.secret();
        let hash = link.hash();

        tracing::debug!("_get_node_from_blobs: Checking for node at hash {}", hash);

        match blobs.stat(hash).await {
            Ok(true) => {
                tracing::debug!("_get_node_from_blobs: Node hash {} exists in blobs", hash);
            }
            Ok(false) => {
                tracing::error!(
                    "_get_node_from_blobs: Node hash {} NOT FOUND in blobs - LinkNotFound error!",
                    hash
                );
                return Err(MountError::LinkNotFound(link.clone()));
            }
            Err(e) => {
                tracing::error!(
                    "_get_node_from_blobs: Error checking node hash {}: {}",
                    hash,
                    e
                );
                return Err(e.into());
            }
        }

        tracing::debug!("_get_node_from_blobs: Reading encrypted node blob");
        let blob = blobs.get(hash).await?;
        tracing::debug!(
            "_get_node_from_blobs: Got {} bytes of encrypted node data",
            blob.len()
        );

        tracing::debug!("_get_node_from_blobs: Decrypting node data");
        let data = secret.decrypt(&blob)?;
        tracing::debug!("_get_node_from_blobs: Decrypted {} bytes", data.len());

        let node = Node::decode(&data)?;
        tracing::debug!("_get_node_from_blobs: Successfully decoded Node");

        Ok(node)
    }

    // TODO (amiller68): you should inline a Link
    //  into the node when we store encrypt it,
    //  so that we have an integrity check
    async fn _put_node_in_blobs(
        node: &Node,
        secret: &Secret,
        blobs: &BlobsStore,
    ) -> Result<Link, MountError> {
        let _data = node.encode()?;
        let data = secret.encrypt(&_data)?;
        let hash = blobs.put(data).await?;
        // NOTE (amiller68): nodes are always stored as raw
        //  since they are encrypted blobs
        let link = Link::new(
            crate::linked_data::LD_RAW_CODEC,
            hash,
            iroh_blobs::BlobFormat::Raw,
        );
        Ok(link)
    }

    pub async fn _put_manifest_in_blobs(
        bucket_data: &Manifest,
        blobs: &BlobsStore,
    ) -> Result<Link, MountError> {
        let data = bucket_data.encode()?;
        let hash = blobs.put(data).await?;
        // NOTE (amiller68): buckets are unencrypted, so they can inherit
        //  the codec of the bucket itself (which is currently always cbor)
        let link = Link::new(bucket_data.codec(), hash, iroh_blobs::BlobFormat::Raw);
        Ok(link)
    }

    pub async fn _put_pins_in_blobs(pins: &Pins, blobs: &BlobsStore) -> Result<Link, MountError> {
        // Create a hash list blob from the pins (raw bytes: 32 bytes per hash, concatenated)
        let hash = blobs.create_hash_list(pins.iter().copied()).await?;
        // Pins are stored as raw blobs containing concatenated hashes
        let link = Link::new(
            crate::linked_data::LD_RAW_CODEC,
            hash,
            iroh_blobs::BlobFormat::HashSeq,
        );
        Ok(link)
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use std::io::Cursor;
    use tempfile::TempDir;

    async fn setup_test_env() -> (Mount, BlobsStore, crate::crypto::SecretKey, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let blob_path = temp_dir.path().join("blobs");

        let secret_key = SecretKey::generate();
        // Generate iroh secret key from random bytes
        let blobs = BlobsStore::load(&blob_path).await.unwrap();

        let mount = Mount::init(Uuid::new_v4(), "test".to_string(), &secret_key, &blobs)
            .await
            .unwrap();

        (mount, blobs, secret_key, temp_dir)
    }

    #[tokio::test]
    async fn test_add_and_cat() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        let data = b"Hello, world!";
        let path = PathBuf::from("/test.txt");

        mount
            .add(&path, Cursor::new(data.to_vec()), &blobs)
            .await
            .unwrap();

        let result = mount.cat(&path, &blobs).await.unwrap();
        assert_eq!(result, data);
    }

    #[tokio::test]
    async fn test_add_with_metadata() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        let data = b"{ \"key\": \"value\" }";
        let path = PathBuf::from("/data.json");

        mount
            .add(&path, Cursor::new(data.to_vec()), &blobs)
            .await
            .unwrap();

        let items = mount.ls(&PathBuf::from("/"), &blobs).await.unwrap();
        assert_eq!(items.len(), 1);

        let (file_path, link) = items.iter().next().unwrap();
        assert_eq!(file_path, &PathBuf::from("data.json"));

        if let Some(data_info) = link.data() {
            assert!(data_info.mime().is_some());
            assert_eq!(data_info.mime().unwrap().as_ref(), "application/json");
        } else {
            panic!("Expected data link with metadata");
        }
    }

    #[tokio::test]
    async fn test_ls() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        mount
            .add(
                &PathBuf::from("/file1.txt"),
                Cursor::new(b"data1".to_vec()),
                &blobs,
            )
            .await
            .unwrap();
        mount
            .add(
                &PathBuf::from("/file2.txt"),
                Cursor::new(b"data2".to_vec()),
                &blobs,
            )
            .await
            .unwrap();
        mount
            .add(
                &PathBuf::from("/dir/file3.txt"),
                Cursor::new(b"data3".to_vec()),
                &blobs,
            )
            .await
            .unwrap();

        let items = mount.ls(&PathBuf::from("/"), &blobs).await.unwrap();
        assert_eq!(items.len(), 3);

        assert!(items.contains_key(&PathBuf::from("file1.txt")));
        assert!(items.contains_key(&PathBuf::from("file2.txt")));
        assert!(items.contains_key(&PathBuf::from("dir")));

        let sub_items = mount.ls(&PathBuf::from("/dir"), &blobs).await.unwrap();
        assert_eq!(sub_items.len(), 1);
        assert!(sub_items.contains_key(&PathBuf::from("dir/file3.txt")));
    }

    #[tokio::test]
    async fn test_ls_deep() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        mount
            .add(&PathBuf::from("/a.txt"), Cursor::new(b"a".to_vec()), &blobs)
            .await
            .unwrap();
        mount
            .add(
                &PathBuf::from("/dir1/b.txt"),
                Cursor::new(b"b".to_vec()),
                &blobs,
            )
            .await
            .unwrap();
        mount
            .add(
                &PathBuf::from("/dir1/dir2/c.txt"),
                Cursor::new(b"c".to_vec()),
                &blobs,
            )
            .await
            .unwrap();
        mount
            .add(
                &PathBuf::from("/dir1/dir2/dir3/d.txt"),
                Cursor::new(b"d".to_vec()),
                &blobs,
            )
            .await
            .unwrap();

        let all_items = mount.ls_deep(&PathBuf::from("/"), &blobs).await.unwrap();

        assert!(all_items.contains_key(&PathBuf::from("a.txt")));
        assert!(all_items.contains_key(&PathBuf::from("dir1")));
        assert!(all_items.contains_key(&PathBuf::from("dir1/b.txt")));
        assert!(all_items.contains_key(&PathBuf::from("dir1/dir2")));
        assert!(all_items.contains_key(&PathBuf::from("dir1/dir2/c.txt")));
        assert!(all_items.contains_key(&PathBuf::from("dir1/dir2/dir3")));
        assert!(all_items.contains_key(&PathBuf::from("dir1/dir2/dir3/d.txt")));
    }

    #[tokio::test]
    async fn test_rm() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        mount
            .add(
                &PathBuf::from("/file1.txt"),
                Cursor::new(b"data1".to_vec()),
                &blobs,
            )
            .await
            .unwrap();
        mount
            .add(
                &PathBuf::from("/file2.txt"),
                Cursor::new(b"data2".to_vec()),
                &blobs,
            )
            .await
            .unwrap();

        let items = mount.ls(&PathBuf::from("/"), &blobs).await.unwrap();
        assert_eq!(items.len(), 2);

        mount
            .rm(&PathBuf::from("/file1.txt"), &blobs)
            .await
            .unwrap();

        let items = mount.ls(&PathBuf::from("/"), &blobs).await.unwrap();
        assert_eq!(items.len(), 1);
        assert!(items.contains_key(&PathBuf::from("file2.txt")));
        assert!(!items.contains_key(&PathBuf::from("file1.txt")));

        let result = mount.cat(&PathBuf::from("/file1.txt"), &blobs).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_nested_operations() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        let files = vec![
            ("/root.txt", b"root" as &[u8]),
            ("/docs/readme.md", b"readme" as &[u8]),
            ("/docs/guide.pdf", b"guide" as &[u8]),
            ("/src/main.rs", b"main" as &[u8]),
            ("/src/lib.rs", b"lib" as &[u8]),
            ("/src/tests/unit.rs", b"unit" as &[u8]),
            ("/src/tests/integration.rs", b"integration" as &[u8]),
        ];

        for (path, data) in &files {
            mount
                .add(&PathBuf::from(path), Cursor::new(data.to_vec()), &blobs)
                .await
                .unwrap();
        }

        for (path, expected_data) in &files {
            let data = mount.cat(&PathBuf::from(path), &blobs).await.unwrap();
            assert_eq!(data, expected_data.to_vec());
        }

        mount
            .rm(&PathBuf::from("/src/tests/unit.rs"), &blobs)
            .await
            .unwrap();

        let result = mount
            .cat(&PathBuf::from("/src/tests/unit.rs"), &blobs)
            .await;
        assert!(result.is_err());

        let data = mount
            .cat(&PathBuf::from("/src/tests/integration.rs"), &blobs)
            .await
            .unwrap();
        assert_eq!(data, b"integration");
    }

    #[tokio::test]
    async fn test_various_file_types() {
        let (mut mount, blobs, _, _temp) = setup_test_env().await;

        let test_files = vec![
            ("/image.png", "image/png"),
            ("/video.mp4", "video/mp4"),
            ("/style.css", "text/css"),
            ("/script.js", "application/javascript"),
            ("/data.json", "application/json"),
            ("/archive.zip", "application/zip"),
            ("/document.pdf", "application/pdf"),
            ("/code.rs", "text/rust"),
        ];

        for (path, expected_mime) in test_files {
            mount
                .add(&PathBuf::from(path), Cursor::new(b"test".to_vec()), &blobs)
                .await
                .unwrap();

            let items = mount.ls(&PathBuf::from("/"), &blobs).await.unwrap();
            let link = items.values().find(|l| l.is_data()).unwrap();

            if let Some(data_info) = link.data() {
                assert!(data_info.mime().is_some());
                assert_eq!(data_info.mime().unwrap().as_ref(), expected_mime);
            }

            mount.rm(&PathBuf::from(path), &blobs).await.unwrap();
        }
    }

    #[tokio::test]
    async fn test_error_cases() {
        let (mount, blobs, _, _temp) = setup_test_env().await;

        let result = mount
            .cat(&PathBuf::from("/does_not_exist.txt"), &blobs)
            .await;
        assert!(result.is_err());

        let result = mount.ls(&PathBuf::from("/does_not_exist"), &blobs).await;
        assert!(result.is_err() || result.unwrap().is_empty());

        let (mut mount, blobs, _, _temp) = setup_test_env().await;
        mount
            .add(
                &PathBuf::from("/dir/file.txt"),
                Cursor::new(b"data".to_vec()),
                &blobs,
            )
            .await
            .unwrap();

        let result = mount.cat(&PathBuf::from("/dir"), &blobs).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_save_load() {
        let (mount, blobs, secret_key, _temp) = setup_test_env().await;
        let link = mount.save(&blobs).await.unwrap();
        let _mount = Mount::load(&link, &secret_key, &blobs).await.unwrap();
    }
}