lcpfs 2026.1.102

//! MDS - Metadata Server
//!
//! This module implements the Metadata Server (MDS), which manages the
//! distributed filesystem namespace. The MDS handles directory operations,
//! file creation, and inode allocation while OSDs handle actual data storage.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────┐
//! │                     Metadata Server                          │
//! │                                                              │
//! │  ┌──────────────────────────────────────────────────────┐  │
//! │  │                   Namespace Tree                       │  │
//! │  │   /                                                    │  │
//! │  │   ├── home/                                            │  │
//! │  │   │   ├── user1/                                       │  │
//! │  │   │   │   ├── file1.txt → inode 1001                  │  │
//! │  │   │   │   └── file2.txt → inode 1002                  │  │
//! │  │   │   └── user2/                                       │  │
//! │  │   └── data/                                            │  │
//! │  │       └── backup.tar → inode 2001                      │  │
//! │  └──────────────────────────────────────────────────────┘  │
//! │                          │                                   │
//! │                          ▼                                   │
//! │  ┌──────────────────────────────────────────────────────┐  │
//! │  │                   Inode Table                          │  │
//! │  │   inode 1001 → pool 0, PG 42, size 1024               │  │
//! │  │   inode 1002 → pool 0, PG 17, size 2048               │  │
//! │  │   inode 2001 → pool 0, PG 99, size 1048576            │  │
//! │  └──────────────────────────────────────────────────────┘  │
//! └─────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Features
//!
//! - Hierarchical namespace (POSIX-like paths)
//! - Inode allocation and management
//! - Directory caching for performance
//! - Capability-based access control
//! - Session management for clients
//! - Distributed MDS (multiple MDS for scalability)

#![cfg_attr(not(feature = "std"), no_std)]

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::format;
use alloc::string::{String, ToString};
use alloc::vec;
use alloc::vec::Vec;
use core::fmt;

#[cfg(feature = "std")]
use std::sync::atomic::{AtomicU64, Ordering};

#[cfg(not(feature = "std"))]
use core::sync::atomic::{AtomicU64, Ordering};

// ============================================================================
// Inode Types
// ============================================================================

/// Inode number type.
pub type InodeNum = u64;

/// Root directory inode number.
pub const INODE_ROOT: InodeNum = 1;
/// Invalid/unallocated inode number.
pub const INODE_INVALID: InodeNum = 0;

/// File types.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FileType {
    /// Regular file.
    Regular,
    /// Directory.
    Directory,
    /// Symbolic link.
    Symlink,
    /// Block device.
    BlockDevice,
    /// Character device.
    CharDevice,
    /// Named pipe (FIFO).
    Fifo,
    /// Unix socket.
    Socket,
}

impl FileType {
    /// Get mode bits for this file type.
    pub fn mode_bits(&self) -> u32 {
        match self {
            FileType::Regular => 0o100000,
            FileType::Directory => 0o040000,
            FileType::Symlink => 0o120000,
            FileType::BlockDevice => 0o060000,
            FileType::CharDevice => 0o020000,
            FileType::Fifo => 0o010000,
            FileType::Socket => 0o140000,
        }
    }

    /// Parse from mode bits.
    pub fn from_mode(mode: u32) -> Self {
        match mode & 0o170000 {
            0o040000 => FileType::Directory,
            0o120000 => FileType::Symlink,
            0o060000 => FileType::BlockDevice,
            0o020000 => FileType::CharDevice,
            0o010000 => FileType::Fifo,
            0o140000 => FileType::Socket,
            _ => FileType::Regular,
        }
    }
}

/// File attributes.
#[derive(Debug, Clone)]
pub struct FileAttrs {
    /// File type.
    pub file_type: FileType,
    /// Permission mode (rwxrwxrwx).
    pub mode: u32,
    /// Owner user ID.
    pub uid: u32,
    /// Owner group ID.
    pub gid: u32,
    /// File size in bytes.
    pub size: u64,
    /// Access time (Unix timestamp).
    pub atime: u64,
    /// Modification time.
    pub mtime: u64,
    /// Change time (metadata change).
    pub ctime: u64,
    /// Number of hard links.
    pub nlink: u32,
    /// Device ID (for device files).
    pub rdev: u64,
}

impl Default for FileAttrs {
    fn default() -> Self {
        Self {
            file_type: FileType::Regular,
            mode: 0o644,
            uid: 0,
            gid: 0,
            size: 0,
            atime: 0,
            mtime: 0,
            ctime: 0,
            nlink: 1,
            rdev: 0,
        }
    }
}

impl FileAttrs {
    /// Create attributes for a new directory.
    pub fn new_dir(mode: u32, uid: u32, gid: u32) -> Self {
        Self {
            file_type: FileType::Directory,
            mode: mode & 0o7777,
            uid,
            gid,
            nlink: 2, // . and parent's entry
            ..Default::default()
        }
    }

    /// Create attributes for a new file.
    pub fn new_file(mode: u32, uid: u32, gid: u32) -> Self {
        Self {
            file_type: FileType::Regular,
            mode: mode & 0o7777,
            uid,
            gid,
            ..Default::default()
        }
    }

    /// Create attributes for a symlink.
    pub fn new_symlink(uid: u32, gid: u32) -> Self {
        Self {
            file_type: FileType::Symlink,
            mode: 0o777,
            uid,
            gid,
            ..Default::default()
        }
    }

    /// Get full mode including file type.
    pub fn full_mode(&self) -> u32 {
        self.file_type.mode_bits() | (self.mode & 0o7777)
    }

    /// Check if this is a directory.
    pub fn is_dir(&self) -> bool {
        self.file_type == FileType::Directory
    }

    /// Check if this is a regular file.
    pub fn is_file(&self) -> bool {
        self.file_type == FileType::Regular
    }

    /// Check if this is a symlink.
    pub fn is_symlink(&self) -> bool {
        self.file_type == FileType::Symlink
    }
}

// ============================================================================
// Inode and Object Location
// ============================================================================

/// Location of object data in the cluster.
#[derive(Debug, Clone)]
pub struct ObjectLocation {
    /// Pool ID.
    pub pool_id: u64,
    /// Placement group ID.
    pub pgid: u64,
    /// Object ID (usually same as inode for files).
    pub oid: u64,
    /// Stripe unit (for striped files).
    pub stripe_unit: u32,
    /// Stripe count.
    pub stripe_count: u32,
}

impl ObjectLocation {
    /// Create a simple (non-striped) location.
    pub fn simple(pool_id: u64, pgid: u64, oid: u64) -> Self {
        Self {
            pool_id,
            pgid,
            oid,
            stripe_unit: 0,
            stripe_count: 1,
        }
    }
}

/// Inode entry in the MDS.
#[derive(Debug, Clone)]
pub struct Inode {
    /// Inode number.
    pub ino: InodeNum,
    /// File attributes.
    pub attrs: FileAttrs,
    /// Object location (for files).
    pub location: Option<ObjectLocation>,
    /// Symlink target (for symlinks).
    pub symlink_target: Option<String>,
    /// Extended attributes.
    pub xattrs: BTreeMap<String, Vec<u8>>,
    /// Version for cache coherence.
    pub version: u64,
}

impl Inode {
    /// Create a new inode.
    pub fn new(ino: InodeNum, attrs: FileAttrs) -> Self {
        Self {
            ino,
            attrs,
            location: None,
            symlink_target: None,
            xattrs: BTreeMap::new(),
            version: 1,
        }
    }

    /// Create root directory inode.
    pub fn root() -> Self {
        let mut inode = Self::new(INODE_ROOT, FileAttrs::new_dir(0o755, 0, 0));
        inode.attrs.nlink = 2;
        inode
    }

    /// Update modification time.
    pub fn touch(&mut self, mtime: u64) {
        self.attrs.mtime = mtime;
        self.attrs.ctime = mtime;
        self.version += 1;
    }
}

// ============================================================================
// Directory Entry
// ============================================================================

/// Directory entry.
#[derive(Debug, Clone)]
pub struct DirEntry {
    /// Entry name.
    pub name: String,
    /// Inode number.
    pub ino: InodeNum,
    /// File type (cached from inode).
    pub file_type: FileType,
}

impl DirEntry {
    /// Create a new directory entry.
    pub fn new(name: &str, ino: InodeNum, file_type: FileType) -> Self {
        Self {
            name: name.to_string(),
            ino,
            file_type,
        }
    }
}

/// Directory contents cache.
#[derive(Debug, Clone)]
pub struct DirContents {
    /// Parent inode.
    pub parent_ino: InodeNum,
    /// Entries (name -> DirEntry).
    pub entries: BTreeMap<String, DirEntry>,
    /// Version for cache invalidation.
    pub version: u64,
}

impl DirContents {
    /// Create new directory contents.
    pub fn new(parent_ino: InodeNum) -> Self {
        let mut entries = BTreeMap::new();

        // Add . and ..
        entries.insert(
            ".".to_string(),
            DirEntry::new(".", parent_ino, FileType::Directory),
        );
        entries.insert(
            "..".to_string(),
            DirEntry::new("..", parent_ino, FileType::Directory),
        );

        Self {
            parent_ino,
            entries,
            version: 1,
        }
    }

    /// Add an entry.
    pub fn add(&mut self, entry: DirEntry) {
        self.entries.insert(entry.name.clone(), entry);
        self.version += 1;
    }

    /// Remove an entry.
    pub fn remove(&mut self, name: &str) -> Option<DirEntry> {
        if name == "." || name == ".." {
            return None;
        }
        let result = self.entries.remove(name);
        if result.is_some() {
            self.version += 1;
        }
        result
    }

    /// Look up an entry.
    pub fn lookup(&self, name: &str) -> Option<&DirEntry> {
        self.entries.get(name)
    }

    /// Get entry count (excluding . and ..).
    pub fn count(&self) -> usize {
        self.entries.len().saturating_sub(2)
    }

    /// Check if empty (excluding . and ..).
    pub fn is_empty(&self) -> bool {
        self.count() == 0
    }
}

// ============================================================================
// MDS Errors
// ============================================================================

/// Errors from MDS operations.
#[derive(Debug, Clone)]
pub enum MdsError {
    /// File or directory not found.
    NotFound(String),
    /// Already exists.
    AlreadyExists(String),
    /// Not a directory.
    NotDirectory(String),
    /// Is a directory.
    IsDirectory(String),
    /// Directory not empty.
    DirectoryNotEmpty(String),
    /// Permission denied.
    PermissionDenied,
    /// Invalid path.
    InvalidPath(String),
    /// Name too long.
    NameTooLong(usize),
    /// Too many symlink levels.
    SymlinkLoop,
    /// No space left.
    NoSpace,
    /// Inode not found.
    InodeNotFound(InodeNum),
    /// Internal error.
    Internal(String),
}

impl fmt::Display for MdsError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            MdsError::NotFound(path) => write!(f, "Not found: {}", path),
            MdsError::AlreadyExists(path) => write!(f, "Already exists: {}", path),
            MdsError::NotDirectory(path) => write!(f, "Not a directory: {}", path),
            MdsError::IsDirectory(path) => write!(f, "Is a directory: {}", path),
            MdsError::DirectoryNotEmpty(path) => write!(f, "Directory not empty: {}", path),
            MdsError::PermissionDenied => write!(f, "Permission denied"),
            MdsError::InvalidPath(path) => write!(f, "Invalid path: {}", path),
            MdsError::NameTooLong(len) => write!(f, "Name too long: {} bytes", len),
            MdsError::SymlinkLoop => write!(f, "Symlink loop detected"),
            MdsError::NoSpace => write!(f, "No space left"),
            MdsError::InodeNotFound(ino) => write!(f, "Inode {} not found", ino),
            MdsError::Internal(msg) => write!(f, "Internal error: {}", msg),
        }
    }
}

// ============================================================================
// Metadata Server
// ============================================================================

/// Maximum name length.
pub const MAX_NAME_LEN: usize = 255;

/// Maximum symlink resolution depth.
pub const MAX_SYMLINK_DEPTH: usize = 40;

/// MDS configuration.
#[derive(Debug, Clone)]
pub struct MdsConfig {
    /// MDS ID (for distributed MDS).
    pub id: u64,
    /// Default pool for new files.
    pub default_pool: u64,
    /// Number of placement groups.
    pub pg_count: u64,
    /// Enable directory caching.
    pub dir_cache_enabled: bool,
    /// Maximum directory cache entries.
    pub dir_cache_max: usize,
}

impl Default for MdsConfig {
    fn default() -> Self {
        Self {
            id: 0,
            default_pool: 0,
            pg_count: 256,
            dir_cache_enabled: true,
            dir_cache_max: 10000,
        }
    }
}

/// Metadata Server.
pub struct Mds {
    /// Configuration.
    config: MdsConfig,
    /// Inode table.
    inodes: BTreeMap<InodeNum, Inode>,
    /// Directory contents cache.
    dir_cache: BTreeMap<InodeNum, DirContents>,
    /// Next inode number.
    next_ino: AtomicU64,
    /// Root inode.
    root_ino: InodeNum,
}

impl Mds {
    /// Create a new MDS.
    pub fn new(config: MdsConfig) -> Self {
        let mut mds = Self {
            config,
            inodes: BTreeMap::new(),
            dir_cache: BTreeMap::new(),
            next_ino: AtomicU64::new(INODE_ROOT + 1),
            root_ino: INODE_ROOT,
        };

        // Create root directory
        let root = Inode::root();
        mds.inodes.insert(INODE_ROOT, root);
        mds.dir_cache
            .insert(INODE_ROOT, DirContents::new(INODE_ROOT));

        mds
    }

    /// Allocate a new inode number.
    fn allocate_ino(&self) -> InodeNum {
        self.next_ino.fetch_add(1, Ordering::SeqCst)
    }

    /// Calculate PG ID for an inode.
    fn inode_to_pg(&self, ino: InodeNum) -> u64 {
        // Simple hash-based placement
        let hash = ino.wrapping_mul(0x9e3779b97f4a7c15);
        hash % self.config.pg_count
    }

    /// Get an inode.
    pub fn get_inode(&self, ino: InodeNum) -> Option<&Inode> {
        self.inodes.get(&ino)
    }

    /// Get a mutable inode.
    pub fn get_inode_mut(&mut self, ino: InodeNum) -> Option<&mut Inode> {
        self.inodes.get_mut(&ino)
    }

    // ========================================================================
    // Path Resolution
    // ========================================================================

    /// Parse path into components.
    fn parse_path(path: &str) -> Vec<&str> {
        path.split('/').filter(|s| !s.is_empty()).collect()
    }

    /// Resolve path to inode.
    pub fn lookup(&self, path: &str) -> Result<InodeNum, MdsError> {
        self.lookup_with_depth(path, 0)
    }

    /// Resolve path with symlink depth tracking.
    fn lookup_with_depth(&self, path: &str, depth: usize) -> Result<InodeNum, MdsError> {
        if depth > MAX_SYMLINK_DEPTH {
            return Err(MdsError::SymlinkLoop);
        }

        if path.is_empty() || path == "/" {
            return Ok(self.root_ino);
        }

        let components = Self::parse_path(path);
        let mut current = self.root_ino;

        for (i, component) in components.iter().enumerate() {
            if component.len() > MAX_NAME_LEN {
                return Err(MdsError::NameTooLong(component.len()));
            }

            // Get current inode
            let inode = self
                .inodes
                .get(&current)
                .ok_or(MdsError::InodeNotFound(current))?;

            // Handle symlinks
            if inode.attrs.is_symlink() {
                if let Some(target) = &inode.symlink_target {
                    let resolved = if target.starts_with('/') {
                        target.clone()
                    } else {
                        // Relative symlink - resolve from parent
                        let parent_path: String = components[..i].join("/");
                        if parent_path.is_empty() {
                            format!("/{}", target)
                        } else {
                            format!("/{}/{}", parent_path, target)
                        }
                    };
                    return self.lookup_with_depth(&resolved, depth + 1);
                }
            }

            // Must be a directory to traverse
            if !inode.attrs.is_dir() {
                return Err(MdsError::NotDirectory(component.to_string()));
            }

            // Look up in directory
            let dir_contents = self
                .dir_cache
                .get(&current)
                .ok_or(MdsError::InodeNotFound(current))?;

            let entry = dir_contents.lookup(component).ok_or_else(|| {
                let remaining: String = components[i..].join("/");
                MdsError::NotFound(remaining)
            })?;

            current = entry.ino;
        }

        Ok(current)
    }

    /// Look up parent directory and final component.
    fn lookup_parent(&self, path: &str) -> Result<(InodeNum, String), MdsError> {
        let path = path.trim_end_matches('/');
        if path.is_empty() || path == "/" {
            return Err(MdsError::InvalidPath(path.to_string()));
        }

        let components = Self::parse_path(path);
        if components.is_empty() {
            return Err(MdsError::InvalidPath(path.to_string()));
        }

        let name = components.last().unwrap().to_string();
        if name.len() > MAX_NAME_LEN {
            return Err(MdsError::NameTooLong(name.len()));
        }

        if components.len() == 1 {
            return Ok((self.root_ino, name));
        }

        let parent_path: String = components[..components.len() - 1].join("/");
        let parent_ino = self.lookup(&format!("/{}", parent_path))?;

        // Verify parent is a directory
        let parent = self
            .inodes
            .get(&parent_ino)
            .ok_or(MdsError::InodeNotFound(parent_ino))?;
        if !parent.attrs.is_dir() {
            return Err(MdsError::NotDirectory(parent_path));
        }

        Ok((parent_ino, name))
    }

    // ========================================================================
    // File Operations
    // ========================================================================

    /// Create a new file.
    pub fn create(&mut self, path: &str, attrs: FileAttrs) -> Result<InodeNum, MdsError> {
        let (parent_ino, name) = self.lookup_parent(path)?;

        // Check if already exists
        if let Some(dir_contents) = self.dir_cache.get(&parent_ino) {
            if dir_contents.lookup(&name).is_some() {
                return Err(MdsError::AlreadyExists(path.to_string()));
            }
        }

        // Allocate inode
        let ino = self.allocate_ino();
        let pgid = self.inode_to_pg(ino);

        // Create inode with location
        let mut inode = Inode::new(ino, attrs);
        inode.location = Some(ObjectLocation::simple(self.config.default_pool, pgid, ino));

        // Add to inode table
        self.inodes.insert(ino, inode);

        // Add to parent directory
        if let Some(dir_contents) = self.dir_cache.get_mut(&parent_ino) {
            dir_contents.add(DirEntry::new(&name, ino, FileType::Regular));
        }

        // Update parent mtime
        if let Some(parent) = self.inodes.get_mut(&parent_ino) {
            parent.touch(0); // Would use actual time
        }

        Ok(ino)
    }

    /// Create a directory.
    pub fn mkdir(
        &mut self,
        path: &str,
        mode: u32,
        uid: u32,
        gid: u32,
    ) -> Result<InodeNum, MdsError> {
        let (parent_ino, name) = self.lookup_parent(path)?;

        // Check if already exists
        if let Some(dir_contents) = self.dir_cache.get(&parent_ino) {
            if dir_contents.lookup(&name).is_some() {
                return Err(MdsError::AlreadyExists(path.to_string()));
            }
        }

        // Allocate inode
        let ino = self.allocate_ino();

        // Create directory inode
        let mut inode = Inode::new(ino, FileAttrs::new_dir(mode, uid, gid));
        inode.attrs.nlink = 2; // . and parent's entry

        // Add to inode table
        self.inodes.insert(ino, inode);

        // Create directory contents with proper parent
        let mut dir_contents = DirContents::new(ino);
        dir_contents.entries.get_mut("..").unwrap().ino = parent_ino;
        self.dir_cache.insert(ino, dir_contents);

        // Add to parent directory
        if let Some(parent_contents) = self.dir_cache.get_mut(&parent_ino) {
            parent_contents.add(DirEntry::new(&name, ino, FileType::Directory));
        }

        // Update parent nlink and mtime
        if let Some(parent) = self.inodes.get_mut(&parent_ino) {
            parent.attrs.nlink += 1;
            parent.touch(0);
        }

        Ok(ino)
    }

    /// Create a symbolic link.
    pub fn symlink(
        &mut self,
        path: &str,
        target: &str,
        uid: u32,
        gid: u32,
    ) -> Result<InodeNum, MdsError> {
        let (parent_ino, name) = self.lookup_parent(path)?;

        // Check if already exists
        if let Some(dir_contents) = self.dir_cache.get(&parent_ino) {
            if dir_contents.lookup(&name).is_some() {
                return Err(MdsError::AlreadyExists(path.to_string()));
            }
        }

        // Allocate inode
        let ino = self.allocate_ino();

        // Create symlink inode
        let mut inode = Inode::new(ino, FileAttrs::new_symlink(uid, gid));
        inode.symlink_target = Some(target.to_string());
        inode.attrs.size = target.len() as u64;

        // Add to inode table
        self.inodes.insert(ino, inode);

        // Add to parent directory
        if let Some(dir_contents) = self.dir_cache.get_mut(&parent_ino) {
            dir_contents.add(DirEntry::new(&name, ino, FileType::Symlink));
        }

        Ok(ino)
    }

    /// Create a hard link.
    pub fn link(&mut self, target: &str, link_path: &str) -> Result<(), MdsError> {
        let target_ino = self.lookup(target)?;
        let (parent_ino, name) = self.lookup_parent(link_path)?;

        // Get target inode
        let target_inode = self
            .inodes
            .get(&target_ino)
            .ok_or(MdsError::InodeNotFound(target_ino))?;

        // Cannot hard link directories
        if target_inode.attrs.is_dir() {
            return Err(MdsError::IsDirectory(target.to_string()));
        }

        let file_type = target_inode.attrs.file_type;

        // Check if already exists
        if let Some(dir_contents) = self.dir_cache.get(&parent_ino) {
            if dir_contents.lookup(&name).is_some() {
                return Err(MdsError::AlreadyExists(link_path.to_string()));
            }
        }

        // Add to parent directory
        if let Some(dir_contents) = self.dir_cache.get_mut(&parent_ino) {
            dir_contents.add(DirEntry::new(&name, target_ino, file_type));
        }

        // Increment nlink
        if let Some(inode) = self.inodes.get_mut(&target_ino) {
            inode.attrs.nlink += 1;
        }

        Ok(())
    }

    /// Remove a file.
    pub fn unlink(&mut self, path: &str) -> Result<(), MdsError> {
        let ino = self.lookup(path)?;
        let (parent_ino, name) = self.lookup_parent(path)?;

        // Check if it's a directory
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        if inode.attrs.is_dir() {
            return Err(MdsError::IsDirectory(path.to_string()));
        }

        // Remove from parent directory
        if let Some(dir_contents) = self.dir_cache.get_mut(&parent_ino) {
            dir_contents.remove(&name);
        }

        // Decrement nlink
        let should_remove = if let Some(inode) = self.inodes.get_mut(&ino) {
            inode.attrs.nlink = inode.attrs.nlink.saturating_sub(1);
            inode.attrs.nlink == 0
        } else {
            false
        };

        // Remove inode if no more links
        if should_remove {
            self.inodes.remove(&ino);
        }

        // Update parent mtime
        if let Some(parent) = self.inodes.get_mut(&parent_ino) {
            parent.touch(0);
        }

        Ok(())
    }

    /// Remove a directory.
    pub fn rmdir(&mut self, path: &str) -> Result<(), MdsError> {
        let ino = self.lookup(path)?;
        let (parent_ino, name) = self.lookup_parent(path)?;

        // Cannot remove root
        if ino == self.root_ino {
            return Err(MdsError::PermissionDenied);
        }

        // Check if it's a directory
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        if !inode.attrs.is_dir() {
            return Err(MdsError::NotDirectory(path.to_string()));
        }

        // Check if empty
        if let Some(dir_contents) = self.dir_cache.get(&ino) {
            if !dir_contents.is_empty() {
                return Err(MdsError::DirectoryNotEmpty(path.to_string()));
            }
        }

        // Remove from parent directory
        if let Some(parent_contents) = self.dir_cache.get_mut(&parent_ino) {
            parent_contents.remove(&name);
        }

        // Remove directory contents cache
        self.dir_cache.remove(&ino);

        // Remove inode
        self.inodes.remove(&ino);

        // Decrement parent nlink
        if let Some(parent) = self.inodes.get_mut(&parent_ino) {
            parent.attrs.nlink = parent.attrs.nlink.saturating_sub(1);
            parent.touch(0);
        }

        Ok(())
    }

    /// Rename a file or directory.
    pub fn rename(&mut self, old_path: &str, new_path: &str) -> Result<(), MdsError> {
        let ino = self.lookup(old_path)?;
        let (old_parent, old_name) = self.lookup_parent(old_path)?;
        let (new_parent, new_name) = self.lookup_parent(new_path)?;

        // Get file type
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        let file_type = inode.attrs.file_type;
        let is_dir = inode.attrs.is_dir();

        // Check if new path already exists
        if let Some(dir_contents) = self.dir_cache.get(&new_parent) {
            if let Some(existing) = dir_contents.lookup(&new_name) {
                if existing.ino != ino {
                    // Would need to handle overwrite
                    return Err(MdsError::AlreadyExists(new_path.to_string()));
                }
            }
        }

        // Remove from old parent
        if let Some(dir_contents) = self.dir_cache.get_mut(&old_parent) {
            dir_contents.remove(&old_name);
        }

        // Add to new parent
        if let Some(dir_contents) = self.dir_cache.get_mut(&new_parent) {
            dir_contents.add(DirEntry::new(&new_name, ino, file_type));
        }

        // Update .. in moved directory
        if is_dir && old_parent != new_parent {
            if let Some(dir_contents) = self.dir_cache.get_mut(&ino) {
                if let Some(dotdot) = dir_contents.entries.get_mut("..") {
                    dotdot.ino = new_parent;
                }
            }

            // Update nlinks
            if let Some(old_parent_inode) = self.inodes.get_mut(&old_parent) {
                old_parent_inode.attrs.nlink = old_parent_inode.attrs.nlink.saturating_sub(1);
            }
            if let Some(new_parent_inode) = self.inodes.get_mut(&new_parent) {
                new_parent_inode.attrs.nlink += 1;
            }
        }

        // Update mtimes
        if let Some(parent) = self.inodes.get_mut(&old_parent) {
            parent.touch(0);
        }
        if old_parent != new_parent {
            if let Some(parent) = self.inodes.get_mut(&new_parent) {
                parent.touch(0);
            }
        }

        Ok(())
    }

    // ========================================================================
    // Directory Operations
    // ========================================================================

    /// Read directory contents.
    pub fn readdir(&self, path: &str) -> Result<Vec<DirEntry>, MdsError> {
        let ino = self.lookup(path)?;

        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        if !inode.attrs.is_dir() {
            return Err(MdsError::NotDirectory(path.to_string()));
        }

        let dir_contents = self
            .dir_cache
            .get(&ino)
            .ok_or(MdsError::InodeNotFound(ino))?;

        Ok(dir_contents.entries.values().cloned().collect())
    }

    /// Read directory with inode (for readdir by inode).
    pub fn readdir_ino(&self, ino: InodeNum) -> Result<Vec<DirEntry>, MdsError> {
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        if !inode.attrs.is_dir() {
            return Err(MdsError::NotDirectory(ino.to_string()));
        }

        let dir_contents = self
            .dir_cache
            .get(&ino)
            .ok_or(MdsError::InodeNotFound(ino))?;

        Ok(dir_contents.entries.values().cloned().collect())
    }

    // ========================================================================
    // Attribute Operations
    // ========================================================================

    /// Get file attributes.
    pub fn getattr(&self, path: &str) -> Result<FileAttrs, MdsError> {
        let ino = self.lookup(path)?;
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        Ok(inode.attrs.clone())
    }

    /// Get attributes by inode.
    pub fn getattr_ino(&self, ino: InodeNum) -> Result<FileAttrs, MdsError> {
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        Ok(inode.attrs.clone())
    }

    /// Set file attributes.
    pub fn setattr(&mut self, path: &str, attrs: FileAttrs) -> Result<(), MdsError> {
        let ino = self.lookup(path)?;
        let inode = self
            .inodes
            .get_mut(&ino)
            .ok_or(MdsError::InodeNotFound(ino))?;

        // Update allowed attributes
        inode.attrs.mode = attrs.mode;
        inode.attrs.uid = attrs.uid;
        inode.attrs.gid = attrs.gid;
        inode.attrs.size = attrs.size;
        inode.attrs.atime = attrs.atime;
        inode.attrs.mtime = attrs.mtime;
        inode.version += 1;

        Ok(())
    }

    /// Update file size (after write).
    pub fn update_size(&mut self, ino: InodeNum, new_size: u64) {
        if let Some(inode) = self.inodes.get_mut(&ino) {
            if new_size > inode.attrs.size {
                inode.attrs.size = new_size;
            }
            inode.version += 1;
        }
    }

    // ========================================================================
    // Extended Attributes
    // ========================================================================

    /// Set extended attribute.
    pub fn setxattr(&mut self, path: &str, name: &str, value: Vec<u8>) -> Result<(), MdsError> {
        let ino = self.lookup(path)?;
        let inode = self
            .inodes
            .get_mut(&ino)
            .ok_or(MdsError::InodeNotFound(ino))?;
        inode.xattrs.insert(name.to_string(), value);
        inode.version += 1;
        Ok(())
    }

    /// Get extended attribute.
    pub fn getxattr(&self, path: &str, name: &str) -> Result<Vec<u8>, MdsError> {
        let ino = self.lookup(path)?;
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        inode
            .xattrs
            .get(name)
            .cloned()
            .ok_or(MdsError::NotFound(name.to_string()))
    }

    /// List extended attributes.
    pub fn listxattr(&self, path: &str) -> Result<Vec<String>, MdsError> {
        let ino = self.lookup(path)?;
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        Ok(inode.xattrs.keys().cloned().collect())
    }

    /// Remove extended attribute.
    pub fn removexattr(&mut self, path: &str, name: &str) -> Result<(), MdsError> {
        let ino = self.lookup(path)?;
        let inode = self
            .inodes
            .get_mut(&ino)
            .ok_or(MdsError::InodeNotFound(ino))?;
        inode.xattrs.remove(name);
        inode.version += 1;
        Ok(())
    }

    // ========================================================================
    // Utility Methods
    // ========================================================================

    /// Get object location for a file.
    pub fn get_location(&self, path: &str) -> Result<ObjectLocation, MdsError> {
        let ino = self.lookup(path)?;
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        inode
            .location
            .clone()
            .ok_or(MdsError::NotFound(path.to_string()))
    }

    /// Get object location by inode.
    pub fn get_location_ino(&self, ino: InodeNum) -> Result<ObjectLocation, MdsError> {
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;
        inode.location.clone().ok_or(MdsError::InodeNotFound(ino))
    }

    /// Read symlink target.
    pub fn readlink(&self, path: &str) -> Result<String, MdsError> {
        let ino = self.lookup(path)?;
        let inode = self.inodes.get(&ino).ok_or(MdsError::InodeNotFound(ino))?;

        if !inode.attrs.is_symlink() {
            return Err(MdsError::InvalidPath(path.to_string()));
        }

        inode
            .symlink_target
            .clone()
            .ok_or(MdsError::NotFound(path.to_string()))
    }

    /// Get statistics.
    pub fn stats(&self) -> MdsStats {
        MdsStats {
            inodes: self.inodes.len() as u64,
            directories: self.dir_cache.len() as u64,
            next_ino: self.next_ino.load(Ordering::SeqCst),
        }
    }
}

/// MDS statistics.
#[derive(Debug, Clone)]
pub struct MdsStats {
    /// Total inodes.
    pub inodes: u64,
    /// Cached directories.
    pub directories: u64,
    /// Next inode number.
    pub next_ino: u64,
}

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

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    fn create_test_mds() -> Mds {
        Mds::new(MdsConfig::default())
    }

    #[test]
    fn test_mds_creation() {
        let mds = create_test_mds();
        assert!(mds.get_inode(INODE_ROOT).is_some());
        assert!(mds.lookup("/").is_ok());
    }

    #[test]
    fn test_create_file() {
        let mut mds = create_test_mds();

        let ino = mds
            .create("/test.txt", FileAttrs::new_file(0o644, 1000, 1000))
            .unwrap();
        assert!(ino > INODE_ROOT);

        let resolved = mds.lookup("/test.txt").unwrap();
        assert_eq!(resolved, ino);
    }

    #[test]
    fn test_mkdir() {
        let mut mds = create_test_mds();

        let ino = mds.mkdir("/subdir", 0o755, 1000, 1000).unwrap();
        assert!(ino > INODE_ROOT);

        let attrs = mds.getattr("/subdir").unwrap();
        assert!(attrs.is_dir());
        assert_eq!(attrs.mode & 0o777, 0o755);
    }

    #[test]
    fn test_nested_directories() {
        let mut mds = create_test_mds();

        mds.mkdir("/a", 0o755, 0, 0).unwrap();
        mds.mkdir("/a/b", 0o755, 0, 0).unwrap();
        mds.mkdir("/a/b/c", 0o755, 0, 0).unwrap();

        let ino = mds
            .create("/a/b/c/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        let resolved = mds.lookup("/a/b/c/file.txt").unwrap();
        assert_eq!(resolved, ino);
    }

    #[test]
    fn test_unlink() {
        let mut mds = create_test_mds();

        mds.create("/test.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        assert!(mds.lookup("/test.txt").is_ok());

        mds.unlink("/test.txt").unwrap();
        assert!(mds.lookup("/test.txt").is_err());
    }

    #[test]
    fn test_rmdir() {
        let mut mds = create_test_mds();

        mds.mkdir("/emptydir", 0o755, 0, 0).unwrap();
        mds.rmdir("/emptydir").unwrap();
        assert!(mds.lookup("/emptydir").is_err());
    }

    #[test]
    fn test_rmdir_not_empty() {
        let mut mds = create_test_mds();

        mds.mkdir("/notempty", 0o755, 0, 0).unwrap();
        mds.create("/notempty/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();

        let result = mds.rmdir("/notempty");
        assert!(matches!(result, Err(MdsError::DirectoryNotEmpty(_))));
    }

    #[test]
    fn test_rename() {
        let mut mds = create_test_mds();

        let ino = mds
            .create("/old.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.rename("/old.txt", "/new.txt").unwrap();

        assert!(mds.lookup("/old.txt").is_err());
        assert_eq!(mds.lookup("/new.txt").unwrap(), ino);
    }

    #[test]
    fn test_rename_across_dirs() {
        let mut mds = create_test_mds();

        mds.mkdir("/src", 0o755, 0, 0).unwrap();
        mds.mkdir("/dst", 0o755, 0, 0).unwrap();

        let ino = mds
            .create("/src/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.rename("/src/file.txt", "/dst/file.txt").unwrap();

        assert!(mds.lookup("/src/file.txt").is_err());
        assert_eq!(mds.lookup("/dst/file.txt").unwrap(), ino);
    }

    #[test]
    fn test_symlink() {
        let mut mds = create_test_mds();

        mds.create("/target.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.symlink("/link.txt", "/target.txt", 0, 0).unwrap();

        let link_attrs = mds.getattr("/link.txt").unwrap();
        assert!(link_attrs.is_symlink());

        let target = mds.readlink("/link.txt").unwrap();
        assert_eq!(target, "/target.txt");
    }

    #[test]
    fn test_hard_link() {
        let mut mds = create_test_mds();

        let ino = mds
            .create("/original.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.link("/original.txt", "/hardlink.txt").unwrap();

        let link_ino = mds.lookup("/hardlink.txt").unwrap();
        assert_eq!(link_ino, ino);

        let attrs = mds.getattr("/original.txt").unwrap();
        assert_eq!(attrs.nlink, 2);
    }

    #[test]
    fn test_readdir() {
        let mut mds = create_test_mds();

        mds.create("/file1.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.create("/file2.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.mkdir("/subdir", 0o755, 0, 0).unwrap();

        let entries = mds.readdir("/").unwrap();

        // Should have ., .., file1.txt, file2.txt, subdir
        assert_eq!(entries.len(), 5);

        let names: Vec<_> = entries.iter().map(|e| e.name.as_str()).collect();
        assert!(names.contains(&"."));
        assert!(names.contains(&".."));
        assert!(names.contains(&"file1.txt"));
        assert!(names.contains(&"file2.txt"));
        assert!(names.contains(&"subdir"));
    }

    #[test]
    fn test_xattrs() {
        let mut mds = create_test_mds();

        mds.create("/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();

        mds.setxattr("/file.txt", "user.test", b"value".to_vec())
            .unwrap();

        let value = mds.getxattr("/file.txt", "user.test").unwrap();
        assert_eq!(value, b"value");

        let list = mds.listxattr("/file.txt").unwrap();
        assert!(list.contains(&"user.test".to_string()));

        mds.removexattr("/file.txt", "user.test").unwrap();
        assert!(mds.getxattr("/file.txt", "user.test").is_err());
    }

    #[test]
    fn test_get_location() {
        let mut mds = create_test_mds();

        let ino = mds
            .create("/data.bin", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        let location = mds.get_location("/data.bin").unwrap();

        assert_eq!(location.oid, ino);
        assert_eq!(location.pool_id, 0);
    }

    #[test]
    fn test_update_size() {
        let mut mds = create_test_mds();

        let ino = mds
            .create("/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();

        mds.update_size(ino, 1024);
        let attrs = mds.getattr_ino(ino).unwrap();
        assert_eq!(attrs.size, 1024);

        // Size should not decrease
        mds.update_size(ino, 512);
        let attrs = mds.getattr_ino(ino).unwrap();
        assert_eq!(attrs.size, 1024);
    }

    #[test]
    fn test_not_found() {
        let mds = create_test_mds();

        let result = mds.lookup("/nonexistent");
        assert!(matches!(result, Err(MdsError::NotFound(_))));
    }

    #[test]
    fn test_not_directory() {
        let mut mds = create_test_mds();

        mds.create("/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();

        let result = mds.lookup("/file.txt/subpath");
        assert!(matches!(result, Err(MdsError::NotDirectory(_))));
    }

    #[test]
    fn test_already_exists() {
        let mut mds = create_test_mds();

        mds.create("/file.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();

        let result = mds.create("/file.txt", FileAttrs::new_file(0o644, 0, 0));
        assert!(matches!(result, Err(MdsError::AlreadyExists(_))));
    }

    #[test]
    fn test_stats() {
        let mut mds = create_test_mds();

        mds.create("/a.txt", FileAttrs::new_file(0o644, 0, 0))
            .unwrap();
        mds.mkdir("/dir", 0o755, 0, 0).unwrap();

        let stats = mds.stats();
        assert_eq!(stats.inodes, 3); // root + a.txt + dir
        assert_eq!(stats.directories, 2); // root + dir
    }

    #[test]
    fn test_file_type_modes() {
        assert_eq!(FileType::Regular.mode_bits(), 0o100000);
        assert_eq!(FileType::Directory.mode_bits(), 0o040000);
        assert_eq!(FileType::Symlink.mode_bits(), 0o120000);

        assert_eq!(FileType::from_mode(0o100644), FileType::Regular);
        assert_eq!(FileType::from_mode(0o040755), FileType::Directory);
    }
}