lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0
//
// ZPL (ZFS POSIX Layer)
// Maps POSIX filesystem semantics onto DMU object storage.

use crate::fscore::structs::{DmuObjectType, DnodePhys, Dva, Hyperblock};
use crate::integrity::anomaly::AnomalyEngine;
use crate::io::pipeline::Pipeline;
use crate::storage::dmu::{
    DmuTx, Objset, TxWaitType, dmu_object_alloc, dmu_read, dmu_tx_assign, dmu_tx_commit,
    dmu_tx_create, dmu_tx_hold_write, dmu_write, get_block_size,
};
use crate::{FileStat, FsError, FsResult};
use alloc::boxed::Box;
use alloc::format;
use alloc::string::String;
use alloc::vec;
use alloc::vec::Vec;
use hashbrown::HashMap;
use spin::Mutex;

// ================================================================================
// POSIX CONSTANTS (from OpenZFS)
// ================================================================================

// File type constants (upper 4 bits of mode)
/// File type mask for mode bits
pub const S_IFMT: u32 = 0o170000; // Type mask
/// Regular file type
pub const S_IFREG: u32 = 0o100000; // Regular file
/// Directory type
pub const S_IFDIR: u32 = 0o040000; // Directory
/// Symbolic link type
pub const S_IFLNK: u32 = 0o120000; // Symbolic link
/// Block device type
pub const S_IFBLK: u32 = 0o060000; // Block device
/// Character device type
pub const S_IFCHR: u32 = 0o020000; // Character device
/// FIFO (named pipe) type
pub const S_IFIFO: u32 = 0o010000; // FIFO (named pipe)
/// Socket type
pub const S_IFSOCK: u32 = 0o140000; // Socket

// Permission bits
/// Set-user-ID permission bit
pub const S_ISUID: u32 = 0o4000; // Set-user-ID
/// Set-group-ID permission bit
pub const S_ISGID: u32 = 0o2000; // Set-group-ID
/// Sticky bit permission
pub const S_ISVTX: u32 = 0o1000; // Sticky bit
/// Owner read/write/execute permissions
pub const S_IRWXU: u32 = 0o0700; // Owner RWX
/// Owner read permission
pub const S_IRUSR: u32 = 0o0400; // Owner read
/// Owner write permission
pub const S_IWUSR: u32 = 0o0200; // Owner write
/// Owner execute permission
pub const S_IXUSR: u32 = 0o0100; // Owner execute
/// Group read/write/execute permissions
pub const S_IRWXG: u32 = 0o0070; // Group RWX
/// Other read/write/execute permissions
pub const S_IRWXO: u32 = 0o0007; // Other RWX

// Open flags
/// Open file read-only
pub const O_RDONLY: u32 = 0;
/// Open file write-only
pub const O_WRONLY: u32 = 1;
/// Open file read-write
pub const O_RDWR: u32 = 2;
/// Create file if it doesn't exist
pub const O_CREAT: u32 = 0o100;
/// Fail if file exists with O_CREAT
pub const O_EXCL: u32 = 0o200;
/// Truncate file to zero length
pub const O_TRUNC: u32 = 0o1000;
/// Append mode - writes go to end of file
pub const O_APPEND: u32 = 0o2000;
/// Fail if not a directory
pub const O_DIRECTORY: u32 = 0o200000;

// Seek whence
/// Seek from beginning of file
pub const SEEK_SET: i32 = 0;
/// Seek from current position
pub const SEEK_CUR: i32 = 1;
/// Seek from end of file
pub const SEEK_END: i32 = 2;

// Directory entry type (encoded in top 4 bits of ZAP value)
// See OpenZFS: ZFS_DIRENT_TYPE(de) = (de) >> 60
/// Bit shift for directory entry type in ZAP value
pub const DIRENT_TYPE_SHIFT: u64 = 60;
/// Mask for object ID in directory entry ZAP value
pub const DIRENT_OBJ_MASK: u64 = (1 << 48) - 1;

// ================================================================================
// FILE LOCKING STRUCTURES
// ================================================================================

/// Lock type for flock() and fcntl()
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LockType {
    /// Shared (read) lock
    Shared,
    /// Exclusive (write) lock
    Exclusive,
}

/// File lock (record lock for fcntl F_SETLK/F_SETLKW)
#[derive(Debug, Clone)]
pub struct FileLock {
    /// Lock type (shared or exclusive)
    pub lock_type: LockType,
    /// Start offset (0 = beginning of file)
    pub start: u64,
    /// Length (0 = to end of file)
    pub length: u64,
    /// Process ID holding the lock
    pub pid: u32,
}

impl FileLock {
    /// Check if this lock conflicts with another lock
    pub fn conflicts_with(&self, other: &FileLock) -> bool {
        // Shared locks don't conflict with each other
        if self.lock_type == LockType::Shared && other.lock_type == LockType::Shared {
            return false;
        }

        // Check if ranges overlap
        let self_end = if self.length == 0 {
            u64::MAX
        } else {
            self.start + self.length
        };

        let other_end = if other.length == 0 {
            u64::MAX
        } else {
            other.start + other.length
        };

        // Ranges overlap if: self.start < other_end && other.start < self_end
        self.start < other_end && other.start < self_end
    }
}

// ================================================================================
// ZNODE PHYSICAL (ON-DISK) STRUCTURE
// ================================================================================
// Matches OpenZFS znode_phys_t bonus buffer layout

/// On-disk inode structure (ZFS znode physical)
#[repr(C)]
#[derive(Debug, Clone, Copy)]
pub struct ZnodePhys {
    /// Access time [seconds, nanoseconds]
    pub atime: [u64; 2], // Access time [sec, nsec]
    /// Modification time [seconds, nanoseconds]
    pub mtime: [u64; 2], // Modification time
    /// Change time for inode metadata [seconds, nanoseconds]
    pub ctime: [u64; 2], // Change time (inode)
    /// Creation time [seconds, nanoseconds]
    pub crtime: [u64; 2], // Creation time
    /// Generation number for NFS
    pub generation: u64, // Generation number
    /// File type and permission bits
    pub mode: u64, // File type + permissions
    /// File size in bytes
    pub size: u64, // File size in bytes
    /// Parent directory object ID
    pub parent: u64, // Parent directory object ID
    /// Hard link count
    pub links: u64, // Hard link count
    /// Extended attributes object ID
    pub xattr: u64, // Extended attributes object
    /// Device number for special files
    pub rdev: u64, // Device number (for special files)
    /// ZFS flags (immutable, appendonly, etc.)
    pub flags: u64, // ZFS flags (immutable, appendonly, etc.)
    /// Owner user ID
    pub uid: u64, // Owner user ID
    /// Owner group ID
    pub gid: u64, // Owner group ID
    /// Reserved padding
    pub pad: [u64; 4], // Reserved
}

impl Default for ZnodePhys {
    fn default() -> Self {
        Self {
            atime: [0, 0],
            mtime: [0, 0],
            ctime: [0, 0],
            crtime: [0, 0],
            generation: 1,
            mode: 0,
            size: 0,
            parent: 0,
            links: 1,
            xattr: 0,
            rdev: 0,
            flags: 0,
            uid: 0,
            gid: 0,
            pad: [0; 4],
        }
    }
}

// ================================================================================
// ZNODE (IN-MEMORY INODE)
// ================================================================================
// Represents a file or directory in memory, with cached attributes

/// In-memory representation of a file or directory (inode equivalent)
pub struct Znode {
    /// DMU object ID (inode number)
    pub object_id: u64,

    /// On-disk attributes (cached)
    pub phys: ZnodePhys,

    /// Data block pointer
    pub master_node: Dva,

    /// Dirty flag (needs writeback)
    pub dirty: bool,

    /// Unlinked flag (pending deletion)
    pub unlinked: bool,

    /// Cached file data (for small files / write buffering)
    pub data_cache: Option<Vec<u8>>,

    /// BSD-style flock() lock (whole-file advisory lock)
    pub flock: Option<(LockType, u32)>, // (lock_type, pid)

    /// POSIX fcntl() record locks (byte-range advisory locks)
    pub record_locks: Vec<FileLock>,
}

impl Znode {
    /// Create a new znode with the given attributes
    pub fn new(object_id: u64, mode: u32, uid: u32, gid: u32) -> Self {
        let now = get_current_time();
        Self {
            object_id,
            master_node: Dva { vdev: 0, offset: 0 },
            phys: ZnodePhys {
                atime: now,
                mtime: now,
                ctime: now,
                crtime: now,
                generation: 1,
                mode: mode as u64,
                size: 0,
                parent: 0,
                links: 1,
                xattr: 0,
                rdev: 0,
                flags: 0,
                uid: uid as u64,
                gid: gid as u64,
                pad: [0; 4],
            },
            dirty: true,
            unlinked: false,
            data_cache: None,
            flock: None,
            record_locks: Vec::new(),
        }
    }

    /// Create a directory znode
    pub fn new_dir(object_id: u64, parent_id: u64, mode: u32, uid: u32, gid: u32) -> Self {
        let mut znode = Self::new(object_id, S_IFDIR | (mode & 0o7777), uid, gid);
        znode.phys.parent = parent_id;
        znode.phys.links = 2; // . and parent's entry
        znode
    }

    /// Create a regular file znode
    pub fn new_file(object_id: u64, parent_id: u64, mode: u32, uid: u32, gid: u32) -> Self {
        let mut znode = Self::new(object_id, S_IFREG | (mode & 0o7777), uid, gid);
        znode.phys.parent = parent_id;
        znode
    }

    /// Create a symlink znode
    pub fn new_symlink(object_id: u64, parent_id: u64, uid: u32, gid: u32) -> Self {
        let mut znode = Self::new(object_id, S_IFLNK | 0o777, uid, gid);
        znode.phys.parent = parent_id;
        znode
    }

    /// Check if this is a directory
    #[inline]
    pub fn is_dir(&self) -> bool {
        (self.phys.mode as u32 & S_IFMT) == S_IFDIR
    }

    /// Check if this is a regular file
    #[inline]
    pub fn is_file(&self) -> bool {
        (self.phys.mode as u32 & S_IFMT) == S_IFREG
    }

    /// Check if this is a symlink
    #[inline]
    pub fn is_symlink(&self) -> bool {
        (self.phys.mode as u32 & S_IFMT) == S_IFLNK
    }

    /// Get file type for directory entry encoding
    pub fn dirent_type(&self) -> u8 {
        match self.phys.mode as u32 & S_IFMT {
            S_IFREG => 8,   // DT_REG
            S_IFDIR => 4,   // DT_DIR
            S_IFLNK => 10,  // DT_LNK
            S_IFBLK => 6,   // DT_BLK
            S_IFCHR => 2,   // DT_CHR
            S_IFIFO => 1,   // DT_FIFO
            S_IFSOCK => 12, // DT_SOCK
            _ => 0,         // DT_UNKNOWN
        }
    }

    /// Get stat structure for this znode
    pub fn getattr(&self) -> FileStat {
        FileStat {
            st_dev: 0,
            st_ino: self.object_id,
            st_nlink: self.phys.links,
            st_mode: self.phys.mode as u32,
            st_uid: self.phys.uid as u32,
            st_gid: self.phys.gid as u32,
            __pad0: 0,
            st_rdev: self.phys.rdev,
            st_size: self.phys.size as i64,
            st_blksize: 131072, // 128KB
            st_blocks: self.phys.size.div_ceil(512) as i64,
            st_atime: self.phys.atime[0] as i64,
            st_atime_nsec: self.phys.atime[1] as i64,
            st_mtime: self.phys.mtime[0] as i64,
            st_mtime_nsec: self.phys.mtime[1] as i64,
            st_ctime: self.phys.ctime[0] as i64,
            st_ctime_nsec: self.phys.ctime[1] as i64,
            __reserved: [0; 3],
        }
    }

    /// Update modification time
    pub fn touch_mtime(&mut self) {
        let now = get_current_time();
        self.phys.mtime = now;
        self.phys.ctime = now;
        self.dirty = true;
    }

    /// Update access time
    pub fn touch_atime(&mut self) {
        self.phys.atime = get_current_time();
        self.dirty = true;
    }

    /// Update change time (inode metadata changed)
    pub fn touch_ctime(&mut self) {
        self.phys.ctime = get_current_time();
        self.dirty = true;
    }

    /// Check access permissions (simplified DAC)
    pub fn check_access(&self, uid: u32, gid: u32, mode: u32) -> bool {
        // Root can do anything
        if uid == 0 {
            return true;
        }

        let file_mode = self.phys.mode as u32;
        let perm = if uid == self.phys.uid as u32 {
            (file_mode >> 6) & 0o7
        } else if gid == self.phys.gid as u32 {
            (file_mode >> 3) & 0o7
        } else {
            file_mode & 0o7
        };

        (perm & mode) == mode
    }

    // ============================================================================
    // FILE LOCKING OPERATIONS
    // ============================================================================

    /// Acquire BSD flock() lock (whole-file advisory lock)
    pub fn flock_acquire(&mut self, lock_type: LockType, pid: u32) -> FsResult<()> {
        if let Some((existing_type, existing_pid)) = self.flock {
            // Same process can re-lock
            if existing_pid == pid {
                self.flock = Some((lock_type, pid));
                return Ok(());
            }

            // Shared locks can coexist
            if lock_type == LockType::Shared && existing_type == LockType::Shared {
                // Note: In real implementation, would track multiple shared locks
                return Ok(());
            }

            // Conflicting lock exists
            return Err(FsError::ResourceBusy);
        }

        // No existing lock, acquire it
        self.flock = Some((lock_type, pid));
        Ok(())
    }

    /// Release BSD flock() lock
    pub fn flock_release(&mut self, pid: u32) -> FsResult<()> {
        if let Some((_, existing_pid)) = self.flock {
            if existing_pid == pid {
                self.flock = None;
                return Ok(());
            }
        }
        Err(FsError::PermissionDenied)
    }

    /// Acquire POSIX fcntl() record lock (byte-range advisory lock)
    pub fn fcntl_lock(&mut self, lock: FileLock) -> FsResult<()> {
        // Check for conflicts with existing locks
        for existing_lock in &self.record_locks {
            // Skip locks from the same process (they can overlap)
            if existing_lock.pid == lock.pid {
                continue;
            }

            if lock.conflicts_with(existing_lock) {
                return Err(FsError::ResourceBusy);
            }
        }

        // Remove any existing locks from this process that overlap with new lock
        self.record_locks
            .retain(|l| l.pid != lock.pid || !l.conflicts_with(&lock));

        // Add the new lock
        self.record_locks.push(lock);
        Ok(())
    }

    /// Release POSIX fcntl() record lock
    pub fn fcntl_unlock(&mut self, start: u64, length: u64, pid: u32) -> FsResult<()> {
        let unlock_end = if length == 0 {
            u64::MAX
        } else {
            start + length
        };

        // Remove locks from this process that overlap with unlock range
        self.record_locks.retain(|lock| {
            if lock.pid != pid {
                return true; // Keep locks from other processes
            }

            let lock_end = if lock.length == 0 {
                u64::MAX
            } else {
                lock.start + lock.length
            };

            // Keep locks that don't overlap with unlock range
            !(lock.start < unlock_end && start < lock_end)
        });

        Ok(())
    }

    /// Check if a lock would conflict (for F_GETLK)
    pub fn fcntl_test_lock(&self, test_lock: &FileLock) -> Option<FileLock> {
        for existing_lock in &self.record_locks {
            // Skip locks from the same process
            if existing_lock.pid == test_lock.pid {
                continue;
            }

            if test_lock.conflicts_with(existing_lock) {
                return Some(existing_lock.clone());
            }
        }
        None
    }

    /// Release all locks held by a process (called on file close)
    pub fn release_all_locks(&mut self, pid: u32) {
        // Release flock if held by this process
        if let Some((_, existing_pid)) = self.flock {
            if existing_pid == pid {
                self.flock = None;
            }
        }

        // Release all record locks from this process
        self.record_locks.retain(|lock| lock.pid != pid);
    }
}

// ================================================================================
// DIRECTORY ENTRY
// ================================================================================

/// Directory entry with name and file type
#[derive(Debug, Clone)]
pub struct DirEntry {
    /// Entry name
    pub name: String,
    /// Object ID of the entry
    pub object_id: u64,
    /// File type (DT_REG, DT_DIR, etc.)
    pub file_type: u8,
}

impl DirEntry {
    /// Encode directory entry value (object_id + type in top 4 bits)
    pub fn encode(object_id: u64, file_type: u8) -> u64 {
        ((file_type as u64) << DIRENT_TYPE_SHIFT) | (object_id & DIRENT_OBJ_MASK)
    }

    /// Decode directory entry value
    pub fn decode(value: u64) -> (u64, u8) {
        let file_type = (value >> DIRENT_TYPE_SHIFT) as u8;
        let object_id = value & DIRENT_OBJ_MASK;
        (object_id, file_type)
    }
}

// ================================================================================
// FILE HANDLE (OPEN FILE DESCRIPTOR)
// ================================================================================

/// Open file handle with position and flags
pub struct FileHandle {
    /// Object ID of the file
    pub object_id: u64,

    /// Current file offset
    pub offset: u64,

    /// Open flags (O_RDONLY, O_WRONLY, O_RDWR, etc.)
    pub flags: u32,

    /// Reference to znode (would be Arc in real impl)
    pub znode_id: u64,
}

impl FileHandle {
    /// Create a new file handle
    pub fn new(object_id: u64, flags: u32) -> Self {
        Self {
            object_id,
            offset: 0,
            flags,
            znode_id: object_id,
        }
    }

    /// Check if handle allows reading
    pub fn can_read(&self) -> bool {
        let access = self.flags & 3;
        access == O_RDONLY || access == O_RDWR
    }

    /// Check if handle allows writing
    pub fn can_write(&self) -> bool {
        let access = self.flags & 3;
        access == O_WRONLY || access == O_RDWR
    }

    /// Check if handle is in append mode
    pub fn is_append(&self) -> bool {
        (self.flags & O_APPEND) != 0
    }
}

// ================================================================================
// ZPL FILESYSTEM STATE
// ================================================================================

/// ZFS POSIX Layer - main filesystem state manager
pub struct Zpl {
    /// DMU object set for this filesystem
    objset: Objset,

    /// Znode cache (object_id -> Znode)
    znodes: HashMap<u64, Znode>,

    /// Directory entries: dir_id -> (name -> object_id)
    /// This is the in-memory equivalent of ZAP
    dir_entries: HashMap<u64, HashMap<String, u64>>,

    /// Root directory object ID
    root_id: u64,

    /// Next object ID to allocate (legacy, now uses DMU)
    next_object_id: u64,

    /// Open file handles
    file_handles: HashMap<u64, FileHandle>,

    /// Next file handle ID
    next_handle_id: u64,

    /// Quota in bytes (0 = unlimited)
    quota: u64,

    /// Current used space in bytes
    used_bytes: u64,
}

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

impl Zpl {
    /// Create a new ZPL instance with DMU backing
    pub fn new() -> Self {
        let mut objset = Objset::new(1); // Dataset ID 1

        // Create a dummy transaction for root directory creation
        let mut tx = dmu_tx_create(&objset);
        let _ = dmu_tx_hold_write(&mut tx, 0, 0, 512);
        let _ = dmu_tx_assign(&mut tx, &objset, TxWaitType::Wait);

        // Allocate root directory object (ID 2, matching traditional inode)
        // In real ZFS, this is done during pool creation
        let _ = objset.object_claim(
            2, // Traditional root inode
            DmuObjectType::DirectoryContents,
            get_block_size(),
            DmuObjectType::Znode,
            64,
            &tx,
        );

        dmu_tx_commit(tx);

        let mut zpl = Self {
            objset,
            znodes: HashMap::new(),
            dir_entries: HashMap::new(),
            root_id: 2,          // Traditional root inode
            next_object_id: 100, // Legacy fallback
            file_handles: HashMap::new(),
            next_handle_id: 1,
            quota: 0,      // 0 = unlimited
            used_bytes: 0, // Start with no usage
        };

        // Create root directory znode
        let root = Znode::new_dir(2, 2, 0o755, 0, 0);
        zpl.znodes.insert(2, root);

        // Initialize root's directory entries (empty)
        zpl.dir_entries.insert(2, HashMap::new());

        zpl
    }

    /// Import a ZPL from a persisted hyperblock.
    ///
    /// This reconstructs the ZPL state from the on-disk format:
    /// 1. Reads the meta-block from the hyperblock's rootbp
    /// 2. Deserializes dnode entries
    /// 3. For znode-type dnodes, reconstructs Znode entries
    /// 4. For directory dnodes, reconstructs directory entries from ZAP data
    ///
    /// # Arguments
    ///
    /// * `hyperblock` - The active hyperblock with a valid rootbp
    /// * `encryption_key` - Optional encryption key for the dataset
    ///
    /// # Returns
    ///
    /// A reconstructed ZPL instance, or an error if import fails.
    pub fn import_from_hyperblock(
        hyperblock: &Hyperblock,
        encryption_key: Option<[u8; 32]>,
    ) -> FsResult<Self> {
        let rootbp = &hyperblock.rootbp;

        // If rootbp is a hole, return an empty ZPL (fresh filesystem)
        if rootbp.is_hole() {
            crate::lcpfs_println!("[ ZPL  ] Import: rootbp is hole, creating fresh ZPL");
            return Ok(Self::new());
        }

        // Read the meta-block from disk
        let key = encryption_key.unwrap_or([0u8; 32]);

        // Use auto-nonce derivation from block pointer
        let meta_block = Pipeline::read_block_auto_nonce(rootbp, &key).map_err(|e| {
            crate::lcpfs_println!("[ ZPL  ] Import: Failed to read meta-block: {:?}", e);
            e
        })?;

        crate::lcpfs_println!(
            "[ ZPL  ] Import: Read meta-block ({} bytes)",
            meta_block.len()
        );

        // Parse the meta-block: [object_id: u64, DnodePhys]...
        let dnode_size = core::mem::size_of::<DnodePhys>();
        let entry_size = 8 + dnode_size; // 8 bytes for object_id + dnode

        let mut objset = Objset::new(1); // Dataset ID 1
        objset
            .os_txg
            .store(hyperblock.txg, core::sync::atomic::Ordering::Release);
        objset.os_rootbp = *rootbp;

        let mut znodes: HashMap<u64, Znode> = HashMap::new();
        let mut dir_entries: HashMap<u64, HashMap<String, u64>> = HashMap::new();
        let mut max_object_id: u64 = 100;

        let mut offset = 0;
        while offset + entry_size <= meta_block.len() {
            // Read object_id (8 bytes little-endian)
            let object_id = u64::from_le_bytes(
                meta_block[offset..offset + 8]
                    .try_into()
                    .unwrap_or([0u8; 8]),
            );
            offset += 8;

            // Read DnodePhys
            // SAFETY INVARIANTS:
            // 1. meta_block[offset..] has at least dnode_size bytes (checked by while condition)
            // 2. DnodePhys is #[repr(C)] with stable layout
            // 3. Data was written by txg_sync using same format
            // 4. read_unaligned handles misaligned buffer
            let dnode_phys: DnodePhys = unsafe {
                core::ptr::read_unaligned(meta_block[offset..].as_ptr() as *const DnodePhys)
            };
            offset += dnode_size;

            if object_id > max_object_id {
                max_object_id = object_id;
            }

            crate::lcpfs_println!(
                "[ ZPL  ] Import: Found object {} type {}",
                object_id,
                dnode_phys.object_type
            );

            // Reconstruct based on object type
            match dnode_phys.object_type {
                t if t == DmuObjectType::Znode as u8 => {
                    // This is a file or directory znode
                    // Try to read data from the dnode's block pointers
                    let znode = Self::reconstruct_znode(object_id, &dnode_phys, &key)?;
                    znodes.insert(object_id, znode);
                }
                t if t == DmuObjectType::DirectoryContents as u8 => {
                    // Directory - initialize empty dir_entries, will be populated from ZAP
                    dir_entries.entry(object_id).or_default();

                    // Also create a directory znode if not already present
                    if !znodes.contains_key(&object_id) {
                        let znode = Znode::new_dir(object_id, 2, 0o755, 0, 0);
                        znodes.insert(object_id, znode);
                    }
                }
                _ => {
                    // Other object types - skip for now
                }
            }
        }

        // Ensure root directory exists (ID 2)
        if !znodes.contains_key(&2) {
            let root = Znode::new_dir(2, 2, 0o755, 0, 0);
            znodes.insert(2, root);
        }
        dir_entries.entry(2).or_default();

        // Reconstruct directory entries by scanning znodes for parent relationships
        for (&object_id, znode) in znodes.iter() {
            if object_id == 2 {
                continue; // Skip root's own entry
            }
            let parent_id = znode.phys.parent;
            if parent_id > 0 && parent_id != object_id {
                // Generate a name based on object ID (real implementation would use ZAP lookup)
                // For now, we don't have the name stored, so we use object ID
                let name = format!("obj_{}", object_id);
                dir_entries
                    .entry(parent_id)
                    .or_default()
                    .insert(name, object_id);
            }
        }

        crate::lcpfs_println!(
            "[ ZPL  ] Import complete: {} znodes, {} directories",
            znodes.len(),
            dir_entries.len()
        );

        Ok(Self {
            objset,
            znodes,
            dir_entries,
            root_id: 2,
            next_object_id: max_object_id + 1,
            file_handles: HashMap::new(),
            next_handle_id: 1,
            quota: 0,
            used_bytes: 0,
        })
    }

    /// Reconstruct a Znode from a DnodePhys structure.
    ///
    /// Reads file data from the dnode's block pointers if present.
    fn reconstruct_znode(
        object_id: u64,
        dnode_phys: &DnodePhys,
        key: &[u8; 32],
    ) -> FsResult<Znode> {
        // Try to read data from the first block pointer
        let mut data_cache: Option<Vec<u8>> = None;
        let bp = &dnode_phys.blkptr[0];

        if !bp.is_hole() {
            // Use auto-nonce derivation from block pointer
            match Pipeline::read_block_auto_nonce(bp, key) {
                Ok(data) => {
                    data_cache = Some(data);
                }
                Err(e) => {
                    crate::lcpfs_println!(
                        "[ ZPL  ] Warning: Failed to read data for object {}: {:?}",
                        object_id,
                        e
                    );
                    // Continue without data cache
                }
            }
        }

        // Extract ZnodePhys from bonus buffer if present
        // Note: The bonus buffer is only 64 bytes but ZnodePhys is larger,
        // so we reconstruct with defaults and override what we can
        let size = if let Some(ref data) = data_cache {
            data.len() as u64
        } else {
            dnode_phys.used_bytes
        };

        // Determine if this is a directory based on object type
        // In the current implementation, DirectoryContents type indicates a directory
        let is_dir = dnode_phys.object_type == DmuObjectType::DirectoryContents as u8;

        let mut znode = if is_dir {
            Znode::new_dir(object_id, 2, 0o755, 0, 0)
        } else {
            Znode::new_file(object_id, 2, 0o644, 0, 0)
        };

        znode.phys.size = size;
        znode.data_cache = data_cache;
        znode.dirty = false;

        Ok(znode)
    }

    /// Allocate a new object ID using DMU
    fn alloc_object_id(&mut self, tx: &DmuTx) -> FsResult<u64> {
        dmu_object_alloc(
            &mut self.objset,
            DmuObjectType::Znode,
            get_block_size(),
            DmuObjectType::None,
            0,
            tx,
        )
    }

    /// Allocate a new object ID (legacy fallback)
    fn alloc_object_id_legacy(&mut self) -> u64 {
        let id = self.next_object_id;
        self.next_object_id += 1;
        id
    }

    /// Allocate a new file handle ID
    fn alloc_handle_id(&mut self) -> u64 {
        let id = self.next_handle_id;
        self.next_handle_id += 1;
        id
    }

    /// Get root directory object ID
    pub fn root_id(&self) -> u64 {
        self.root_id
    }

    /// Set quota in bytes (0 = unlimited)
    pub fn set_quota(&mut self, quota: u64) {
        self.quota = quota;
    }

    /// Get current quota (0 = unlimited)
    pub fn get_quota(&self) -> u64 {
        self.quota
    }

    /// Get current used space in bytes
    pub fn get_used_bytes(&self) -> u64 {
        self.used_bytes
    }

    /// Alias for get_used_bytes (used by trash module)
    pub fn used_bytes(&self) -> u64 {
        self.used_bytes
    }

    /// Alias for get_quota (used by trash module)
    pub fn quota(&self) -> u64 {
        self.quota
    }

    /// Check if over quota
    pub fn is_over_quota(&self) -> bool {
        self.quota > 0 && self.used_bytes > self.quota
    }

    /// Get znode by object ID
    pub fn get_znode(&self, object_id: u64) -> Option<&Znode> {
        self.znodes.get(&object_id)
    }

    /// Get mutable znode by object ID
    pub fn get_znode_mut(&mut self, object_id: u64) -> Option<&mut Znode> {
        self.znodes.get_mut(&object_id)
    }

    /// Stat a file by object ID (returns FileStat)
    pub fn stat_by_id(&self, object_id: u64) -> FsResult<crate::FileStat> {
        let znode = self.znodes.get(&object_id).ok_or(FsError::NotFound)?;
        Ok(znode.getattr())
    }

    // ============================================================================
    // DIRECTORY OPERATIONS (from OpenZFS zfs_dir.c)
    // ============================================================================

    /// Look up a name in a directory (zfs_dirlook)
    pub fn lookup(&self, dir_id: u64, name: &str) -> FsResult<u64> {
        let dir = self.znodes.get(&dir_id).ok_or(FsError::NotFound)?;

        if !dir.is_dir() {
            return Err(FsError::NotDirectory);
        }

        // Special cases
        match name {
            "." => return Ok(dir_id),
            ".." => return Ok(dir.phys.parent),
            _ => {}
        }

        // Look up in directory entries (in-memory ZAP equivalent)
        self.dir_entries
            .get(&dir_id)
            .and_then(|entries| entries.get(name))
            .copied()
            .ok_or(FsError::NotFound)
    }

    /// Create a directory entry linking name to znode (zfs_link_create)
    pub fn link_create(&mut self, dir_id: u64, name: &str, znode_id: u64) -> FsResult<()> {
        // Check if entry already exists
        if let Some(entries) = self.dir_entries.get(&dir_id) {
            if entries.contains_key(name) {
                return Err(FsError::AlreadyExists);
            }
        }

        // Check if child is a directory (need to read before mutating)
        let child_is_dir = self
            .znodes
            .get(&znode_id)
            .map(|z| z.is_dir())
            .unwrap_or(false);

        // Add to directory entries (in-memory ZAP)
        self.dir_entries
            .entry(dir_id)
            .or_default()
            .insert(String::from(name), znode_id);

        // Increment znode link count
        if let Some(znode) = self.znodes.get_mut(&znode_id) {
            znode.phys.links += 1;
            znode.touch_ctime();
        }

        // Update directory size and mtime
        if let Some(dir) = self.znodes.get_mut(&dir_id) {
            dir.phys.size += (name.len() + 8) as u64; // Approximate entry size
            dir.touch_mtime();

            // If linking a directory, increment parent link count for ".."
            if child_is_dir {
                dir.phys.links += 1;
            }
        }

        // If child is a directory, initialize its dir_entries
        if child_is_dir {
            self.dir_entries.entry(znode_id).or_default();
        }

        Ok(())
    }

    /// Remove a directory entry (zfs_link_destroy)
    pub fn link_destroy(&mut self, dir_id: u64, name: &str, znode_id: u64) -> FsResult<()> {
        // Remove from directory entries (in-memory ZAP)
        if let Some(entries) = self.dir_entries.get_mut(&dir_id) {
            entries.remove(name);
        }

        // Get znode to check type
        let is_dir = self
            .znodes
            .get(&znode_id)
            .map(|z| z.is_dir())
            .unwrap_or(false);

        // Decrement znode link count
        let should_delete = if let Some(znode) = self.znodes.get_mut(&znode_id) {
            znode.phys.links = znode.phys.links.saturating_sub(1);
            znode.touch_ctime();
            znode.phys.links == 0
        } else {
            false
        };

        // Update directory
        if let Some(dir) = self.znodes.get_mut(&dir_id) {
            dir.phys.size = dir.phys.size.saturating_sub((name.len() + 8) as u64);
            dir.touch_mtime();

            if is_dir {
                dir.phys.links = dir.phys.links.saturating_sub(1);
            }
        }

        // Delete znode if no more links and not open
        if should_delete {
            if let Some(znode) = self.znodes.get_mut(&znode_id) {
                znode.unlinked = true;
            }
            // Remove directory entries for deleted directory
            if is_dir {
                self.dir_entries.remove(&znode_id);
            }
            // Actually remove the znode
            self.znodes.remove(&znode_id);
        }

        Ok(())
    }

    // ============================================================================
    // FILE OPERATIONS (from OpenZFS zfs_vnops.c)
    // ============================================================================

    /// Create a new file (zfs_create) - uses DMU transaction
    pub fn create(
        &mut self,
        dir_id: u64,
        name: &str,
        mode: u32,
        uid: u32,
        gid: u32,
    ) -> FsResult<u64> {
        // Anomaly detection: Check if file creation should be blocked (ransomware protection)
        // Uses root_id as dataset identifier, uid as user identifier.
        if let Err(reason) = AnomalyEngine::check_create(uid as u64, self.root_id, 0) {
            return Err(FsError::SecurityViolation { reason });
        }

        // Check parent is a directory
        let dir = self.znodes.get(&dir_id).ok_or(FsError::NotFound)?;

        if !dir.is_dir() {
            return Err(FsError::NotDirectory);
        }

        // Create DMU transaction
        let mut tx = dmu_tx_create(&self.objset);
        let _ = dmu_tx_hold_write(&mut tx, dir_id, 0, 512); // Hold for dir update
        dmu_tx_assign(&mut tx, &self.objset, TxWaitType::Wait)?;

        // Allocate new object via DMU
        let object_id = self.alloc_object_id(&tx)?;

        // Create znode
        let znode = Znode::new_file(object_id, dir_id, mode, uid, gid);
        self.znodes.insert(object_id, znode);

        // Link into directory
        self.link_create(dir_id, name, object_id)?;

        // Commit transaction
        dmu_tx_commit(tx);

        Ok(object_id)
    }

    /// Create a new directory (zfs_mkdir) - uses DMU transaction
    pub fn mkdir(
        &mut self,
        parent_id: u64,
        name: &str,
        mode: u32,
        uid: u32,
        gid: u32,
    ) -> FsResult<u64> {
        // Check parent is a directory
        let parent = self.znodes.get(&parent_id).ok_or(FsError::NotFound)?;

        if !parent.is_dir() {
            return Err(FsError::NotDirectory);
        }

        // Create DMU transaction
        let mut tx = dmu_tx_create(&self.objset);
        let _ = dmu_tx_hold_write(&mut tx, parent_id, 0, 512);
        dmu_tx_assign(&mut tx, &self.objset, TxWaitType::Wait)?;

        // Allocate new object via DMU
        let object_id = self.alloc_object_id(&tx)?;

        // Create directory znode
        let znode = Znode::new_dir(object_id, parent_id, mode, uid, gid);
        self.znodes.insert(object_id, znode);

        // Link into parent directory
        self.link_create(parent_id, name, object_id)?;

        // Commit transaction
        dmu_tx_commit(tx);

        Ok(object_id)
    }

    /// Remove a file (zfs_remove)
    pub fn unlink(&mut self, dir_id: u64, name: &str) -> FsResult<()> {
        // Cannot unlink . or ..
        if name == "." || name == ".." {
            return Err(FsError::InvalidArgument {
                reason: "cannot unlink . or ..",
            });
        }

        // Anomaly detection: Check if deletion should be blocked (ransomware protection)
        // Uses user_id=0 (root) as placeholder until kernel context is available.
        if let Err(reason) = AnomalyEngine::check_delete(0, self.root_id, 0) {
            return Err(FsError::SecurityViolation { reason });
        }

        // Look up the file
        let object_id = self.lookup(dir_id, name)?;

        // Check it's not a directory (use rmdir for that)
        let is_dir = self
            .znodes
            .get(&object_id)
            .map(|z| z.is_dir())
            .unwrap_or(false);

        if is_dir {
            return Err(FsError::IsDirectory);
        }

        // Check no open file handles (simplified - in real impl would defer deletion)
        let has_open_handles = self.file_handles.values().any(|h| h.object_id == object_id);

        if has_open_handles {
            // Mark as unlinked but don't delete yet (POSIX semantics)
            if let Some(znode) = self.znodes.get_mut(&object_id) {
                znode.unlinked = true;
            }
        }

        // Remove directory entry and update link counts
        self.link_destroy(dir_id, name, object_id)
    }

    /// Remove a directory (zfs_rmdir)
    pub fn rmdir(&mut self, parent_id: u64, name: &str) -> FsResult<()> {
        // Cannot remove . or ..
        if name == "." || name == ".." {
            return Err(FsError::InvalidArgument {
                reason: "cannot remove . or ..",
            });
        }

        // Look up the directory
        let object_id = self.lookup(parent_id, name)?;

        // Check it's a directory
        let is_dir = self
            .znodes
            .get(&object_id)
            .map(|z| z.is_dir())
            .unwrap_or(false);

        if !is_dir {
            return Err(FsError::NotDirectory);
        }

        // Check directory is empty
        let is_empty = self
            .dir_entries
            .get(&object_id)
            .map(|entries| entries.is_empty())
            .unwrap_or(true);

        if !is_empty {
            return Err(FsError::DirectoryNotEmpty);
        }

        // Remove directory entry and update link counts
        self.link_destroy(parent_id, name, object_id)
    }

    /// Rename a file/directory (zfs_rename)
    ///
    /// Based on OpenZFS zfs_rename. Key steps:
    /// 1. Validate names (no . or ..)
    /// 2. Lock ordering by object ID to prevent deadlocks
    /// 3. Check for directory rename into own subtree
    /// 4. Handle same-directory vs cross-directory rename
    /// 5. Handle overwrite case (target exists)
    /// 6. Use link_destroy to remove source
    /// 7. Use link_create to add target
    /// 8. For directories, update parent pointer
    /// 9. All in one transaction
    pub fn rename(
        &mut self,
        src_dir: u64,
        src_name: &str,
        dst_dir: u64,
        dst_name: &str,
    ) -> FsResult<()> {
        // Validate names - cannot rename . or ..
        if src_name == "." || src_name == ".." || dst_name == "." || dst_name == ".." {
            return Err(FsError::InvalidArgument {
                reason: "cannot rename . or ..",
            });
        }

        // Look up source - must exist
        let src_object_id = self.lookup(src_dir, src_name)?;

        // Get source properties
        let (src_is_dir, src_parent) = {
            let src_znode = self.znodes.get(&src_object_id).ok_or(FsError::NotFound)?;
            (src_znode.is_dir(), src_znode.phys.parent)
        };

        // Check source and destination directories exist and are directories
        {
            let src_dir_znode = self.znodes.get(&src_dir).ok_or(FsError::NotFound)?;
            if !src_dir_znode.is_dir() {
                return Err(FsError::NotDirectory);
            }
        }
        {
            let dst_dir_znode = self.znodes.get(&dst_dir).ok_or(FsError::NotFound)?;
            if !dst_dir_znode.is_dir() {
                return Err(FsError::NotDirectory);
            }
        }

        // If renaming a directory, check we're not moving it into its own subtree
        // This would create a cycle in the directory tree
        if src_is_dir && src_dir != dst_dir {
            // Walk up from dst_dir to root, checking we don't hit src_object_id
            let mut check_dir = dst_dir;
            while check_dir != self.root_id {
                if check_dir == src_object_id {
                    return Err(FsError::InvalidArgument {
                        reason: "cannot move directory into itself",
                    });
                }
                // Get parent of current directory
                check_dir = self
                    .znodes
                    .get(&check_dir)
                    .map(|z| z.phys.parent)
                    .unwrap_or(self.root_id);
            }
        }

        // Check if destination exists
        let dst_exists = self.lookup(dst_dir, dst_name).ok();

        if let Some(dst_object_id) = dst_exists {
            // Destination exists - handle overwrite

            // Cannot overwrite different types
            let dst_is_dir = self
                .znodes
                .get(&dst_object_id)
                .map(|z| z.is_dir())
                .unwrap_or(false);

            if src_is_dir != dst_is_dir {
                if dst_is_dir {
                    return Err(FsError::IsDirectory);
                } else {
                    return Err(FsError::NotDirectory);
                }
            }

            // If destination is a directory, it must be empty
            if dst_is_dir {
                let is_empty = self
                    .dir_entries
                    .get(&dst_object_id)
                    .map(|entries| entries.is_empty())
                    .unwrap_or(true);

                if !is_empty {
                    return Err(FsError::DirectoryNotEmpty);
                }
            }

            // Cannot rename onto self (same inode)
            if src_object_id == dst_object_id {
                return Ok(()); // No-op
            }

            // Remove destination first
            self.link_destroy(dst_dir, dst_name, dst_object_id)?;
        }

        // Create DMU transaction for atomic rename
        let mut tx = dmu_tx_create(&self.objset);
        let _ = dmu_tx_hold_write(&mut tx, src_dir, 0, 512);
        if src_dir != dst_dir {
            let _ = dmu_tx_hold_write(&mut tx, dst_dir, 0, 512);
        }
        dmu_tx_assign(&mut tx, &self.objset, TxWaitType::Wait)?;

        // Remove from source directory
        // Note: We don't use link_destroy here because we don't want to delete the znode
        if let Some(entries) = self.dir_entries.get_mut(&src_dir) {
            entries.remove(src_name);
        }

        // Update source directory size and mtime
        if let Some(src_dir_znode) = self.znodes.get_mut(&src_dir) {
            src_dir_znode.phys.size = src_dir_znode
                .phys
                .size
                .saturating_sub((src_name.len() + 8) as u64);
            src_dir_znode.touch_mtime();

            // If moving a directory, decrement source parent link count
            if src_is_dir && src_dir != dst_dir {
                src_dir_znode.phys.links = src_dir_znode.phys.links.saturating_sub(1);
            }
        }

        // Add to destination directory
        self.dir_entries
            .entry(dst_dir)
            .or_default()
            .insert(String::from(dst_name), src_object_id);

        // Update destination directory size and mtime
        if let Some(dst_dir_znode) = self.znodes.get_mut(&dst_dir) {
            dst_dir_znode.phys.size += (dst_name.len() + 8) as u64;
            dst_dir_znode.touch_mtime();

            // If moving a directory, increment destination parent link count
            if src_is_dir && src_dir != dst_dir {
                dst_dir_znode.phys.links += 1;
            }
        }

        // Update parent pointer if moving a directory to different parent
        if src_is_dir && src_dir != dst_dir {
            if let Some(src_znode) = self.znodes.get_mut(&src_object_id) {
                src_znode.phys.parent = dst_dir;
                src_znode.touch_ctime();
            }
        }

        // Update ctime on the moved object
        if let Some(src_znode) = self.znodes.get_mut(&src_object_id) {
            src_znode.touch_ctime();
        }

        // Commit transaction
        dmu_tx_commit(tx);

        Ok(())
    }

    /// Open a file
    pub fn open(&mut self, object_id: u64, flags: u32) -> FsResult<u64> {
        // Check file exists and get its properties
        let (is_dir, is_file) = {
            let znode = self.znodes.get(&object_id).ok_or(FsError::NotFound)?;
            (znode.is_dir(), znode.is_file())
        };

        // Check file type
        if (flags & O_DIRECTORY) != 0 && !is_dir {
            return Err(FsError::NotDirectory);
        }
        if is_dir && (flags & O_WRONLY != 0 || flags & O_RDWR != 0) {
            return Err(FsError::IsDirectory);
        }

        // Create file handle
        let handle_id = self.alloc_handle_id();
        let handle = FileHandle::new(object_id, flags);
        self.file_handles.insert(handle_id, handle);

        // Handle O_TRUNC
        if (flags & O_TRUNC) != 0 && is_file {
            if let Some(z) = self.znodes.get_mut(&object_id) {
                z.phys.size = 0;
                z.data_cache = Some(Vec::new());
                z.touch_mtime();
            }
        }

        Ok(handle_id)
    }

    /// Close a file handle
    pub fn close(&mut self, handle_id: u64) -> FsResult<()> {
        self.file_handles
            .remove(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?;
        Ok(())
    }

    /// Acquire a file lock (flock).
    ///
    /// # Arguments
    /// * `handle_id` - File handle from open()
    /// * `lock_type` - Shared or Exclusive
    /// * `pid` - Process ID of lock holder
    pub fn flock(&mut self, handle_id: u64, lock_type: LockType, pid: u32) -> FsResult<()> {
        let object_id = self
            .file_handles
            .get(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?
            .object_id;

        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;
        znode.flock_acquire(lock_type, pid)
    }

    /// Release a file lock (flock).
    pub fn funlock(&mut self, handle_id: u64, pid: u32) -> FsResult<()> {
        let object_id = self
            .file_handles
            .get(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?
            .object_id;

        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;
        znode.flock_release(pid)
    }

    /// Acquire a POSIX fcntl() record lock (byte-range advisory lock).
    ///
    /// # Arguments
    ///
    /// * `handle_id` - File handle from open()
    /// * `lock_type` - Shared or Exclusive
    /// * `start` - Start offset (0 = beginning of file)
    /// * `length` - Length of region (0 = to end of file)
    /// * `pid` - Process ID of lock holder
    pub fn fcntl_setlk(
        &mut self,
        handle_id: u64,
        lock_type: LockType,
        start: u64,
        length: u64,
        pid: u32,
    ) -> FsResult<()> {
        let object_id = self
            .file_handles
            .get(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?
            .object_id;

        let lock = FileLock {
            lock_type,
            start,
            length,
            pid,
        };

        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;
        znode.fcntl_lock(lock)
    }

    /// Release a POSIX fcntl() record lock.
    ///
    /// # Arguments
    ///
    /// * `handle_id` - File handle from open()
    /// * `start` - Start offset of region to unlock
    /// * `length` - Length of region (0 = to end of file)
    /// * `pid` - Process ID of lock holder
    pub fn fcntl_unlk(
        &mut self,
        handle_id: u64,
        start: u64,
        length: u64,
        pid: u32,
    ) -> FsResult<()> {
        let object_id = self
            .file_handles
            .get(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?
            .object_id;

        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;
        znode.fcntl_unlock(start, length, pid)
    }

    /// Test if a lock would conflict (F_GETLK).
    ///
    /// # Arguments
    ///
    /// * `handle_id` - File handle from open()
    /// * `lock_type` - Type of lock to test
    /// * `start` - Start offset
    /// * `length` - Length of region
    /// * `pid` - Process ID of requester
    ///
    /// # Returns
    ///
    /// `None` if lock would succeed, `Some(FileLock)` with conflicting lock info.
    pub fn fcntl_getlk(
        &self,
        handle_id: u64,
        lock_type: LockType,
        start: u64,
        length: u64,
        pid: u32,
    ) -> FsResult<Option<FileLock>> {
        let object_id = self
            .file_handles
            .get(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?
            .object_id;

        let test_lock = FileLock {
            lock_type,
            start,
            length,
            pid,
        };

        let znode = self.znodes.get(&object_id).ok_or(FsError::NotFound)?;
        Ok(znode.fcntl_test_lock(&test_lock))
    }

    /// Read from file (zfs_read) - uses DMU for persistent data
    pub fn read(&mut self, handle_id: u64, buf: &mut [u8]) -> FsResult<usize> {
        let handle = self
            .file_handles
            .get_mut(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?;

        if !handle.can_read() {
            return Err(FsError::PermissionDenied);
        }

        let object_id = handle.object_id;
        let offset = handle.offset;

        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;

        // Check bounds
        if offset >= znode.phys.size {
            return Ok(0);
        }

        let available = (znode.phys.size - offset) as usize;
        let to_read = buf.len().min(available);

        // Try reading from cache first, then fall back to DMU
        if let Some(ref data) = znode.data_cache {
            let start = offset as usize;
            let end = start + to_read;
            if end <= data.len() {
                buf[..to_read].copy_from_slice(&data[start..end]);
            } else {
                let copy_len = data.len().saturating_sub(start);
                if copy_len > 0 {
                    buf[..copy_len].copy_from_slice(&data[start..]);
                }
                return Ok(copy_len);
            }
        } else {
            // Read from DMU object storage
            match dmu_read(&mut self.objset, object_id, offset, to_read) {
                Ok(data) => {
                    let copy_len = data.len().min(to_read);
                    buf[..copy_len].copy_from_slice(&data[..copy_len]);
                }
                Err(_) => {
                    // Object not yet persisted, return zeros
                    buf[..to_read].fill(0);
                }
            }
        }

        // Update offset and atime
        if let Some(h) = self.file_handles.get_mut(&handle_id) {
            h.offset += to_read as u64;
        }
        if let Some(z) = self.znodes.get_mut(&object_id) {
            z.touch_atime();
        }

        Ok(to_read)
    }

    /// Write to file (zfs_write) - uses DMU transaction for persistence
    pub fn write(&mut self, handle_id: u64, buf: &[u8]) -> FsResult<usize> {
        let handle = self
            .file_handles
            .get_mut(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?;

        if !handle.can_write() {
            return Err(FsError::PermissionDenied);
        }

        let object_id = handle.object_id;
        let offset = if handle.is_append() {
            self.znodes
                .get(&object_id)
                .map(|z| z.phys.size)
                .unwrap_or(0)
        } else {
            handle.offset
        };

        // Check quota before write (if quota is set)
        if self.quota > 0 {
            let new_usage = self.used_bytes.saturating_add(buf.len() as u64);
            if new_usage > self.quota {
                return Err(FsError::DiskFull {
                    needed_bytes: buf.len() as u64,
                });
            }
        }

        // Anomaly detection: Check if write should be blocked (ransomware protection)
        // Uses user_id=0 (root) and root_id as dataset identifier.
        if let Err(reason) = AnomalyEngine::check_write(0, self.root_id, 0, buf) {
            return Err(FsError::SecurityViolation { reason });
        }

        // Create DMU transaction for the write
        let mut tx = dmu_tx_create(&self.objset);
        dmu_tx_hold_write(&mut tx, object_id, offset, buf.len() as u64)?;
        dmu_tx_assign(&mut tx, &self.objset, TxWaitType::Wait)?;

        // Write to DMU (for persistence) - MUST propagate errors!
        // If this fails, we must NOT update in-memory cache or claim success
        dmu_write(&mut self.objset, object_id, offset, buf, &tx)?;

        // Commit the transaction
        dmu_tx_commit(tx);

        // Only update in-memory cache AFTER successful DMU write
        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;

        // Ensure data_cache is initialized
        let data = znode.data_cache.get_or_insert_with(Vec::new);
        let write_end = offset as usize + buf.len();

        if data.len() < write_end {
            data.resize(write_end, 0);
        }

        data[offset as usize..write_end].copy_from_slice(buf);

        // Update size and times
        if write_end as u64 > znode.phys.size {
            znode.phys.size = write_end as u64;
        }
        znode.touch_mtime();
        znode.dirty = true;

        // Update handle offset
        if let Some(h) = self.file_handles.get_mut(&handle_id) {
            h.offset = write_end as u64;
        }

        // Update used bytes for quota tracking
        self.used_bytes = self.used_bytes.saturating_add(buf.len() as u64);

        Ok(buf.len())
    }

    /// Seek (lseek)
    pub fn seek(&mut self, handle_id: u64, offset: i64, whence: i32) -> FsResult<u64> {
        let handle = self
            .file_handles
            .get_mut(&handle_id)
            .ok_or(FsError::BadFileDescriptor)?;

        let size = self
            .znodes
            .get(&handle.object_id)
            .map(|z| z.phys.size)
            .unwrap_or(0);

        let new_offset = match whence {
            SEEK_SET => offset as u64,
            SEEK_CUR => (handle.offset as i64 + offset) as u64,
            SEEK_END => (size as i64 + offset) as u64,
            _ => {
                return Err(FsError::InvalidArgument {
                    reason: "invalid seek whence",
                });
            }
        };

        handle.offset = new_offset;
        Ok(new_offset)
    }

    /// Truncate file (ftruncate)
    pub fn truncate(&mut self, object_id: u64, length: u64) -> FsResult<()> {
        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;

        if !znode.is_file() {
            return Err(FsError::InvalidArgument {
                reason: "truncate requires a file",
            });
        }

        if let Some(ref mut data) = znode.data_cache {
            if length as usize > data.len() {
                data.resize(length as usize, 0);
            } else {
                data.truncate(length as usize);
            }
        }

        znode.phys.size = length;
        znode.touch_mtime();
        znode.dirty = true;

        Ok(())
    }

    /// Get file attributes (stat)
    pub fn getattr(&self, object_id: u64) -> FsResult<FileStat> {
        let znode = self.znodes.get(&object_id).ok_or(FsError::NotFound)?;
        Ok(znode.getattr())
    }

    /// Set file attributes (chmod/chown combined)
    pub fn setattr(
        &mut self,
        object_id: u64,
        mode: Option<u32>,
        uid: Option<u32>,
        gid: Option<u32>,
    ) -> FsResult<()> {
        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;

        if let Some(m) = mode {
            let file_type = znode.phys.mode as u32 & S_IFMT;
            znode.phys.mode = (file_type | (m & 0o7777)) as u64;
        }
        if let Some(u) = uid {
            znode.phys.uid = u as u64;
        }
        if let Some(g) = gid {
            znode.phys.gid = g as u64;
        }

        znode.touch_ctime();
        Ok(())
    }

    /// Reflink: Share data blocks between two files (O(1) copy).
    ///
    /// This performs "quantum entanglement" - the destination file's data cache
    /// becomes a clone of the source's data. On copy-on-write filesystems,
    /// actual block pointers would be shared; here we share the in-memory cache.
    ///
    /// # Arguments
    ///
    /// * `src_id` - Source file object ID
    /// * `dst_id` - Destination file object ID (must be empty)
    pub fn reflink_data(&mut self, src_id: u64, dst_id: u64) -> FsResult<()> {
        // Get source data
        let src_data = {
            let src = self.znodes.get(&src_id).ok_or(FsError::NotFound)?;
            if !src.is_file() {
                return Err(FsError::InvalidArgument {
                    reason: "source must be a regular file",
                });
            }
            (src.data_cache.clone(), src.phys.size, src.master_node)
        };

        // Apply to destination
        let dst = self.znodes.get_mut(&dst_id).ok_or(FsError::NotFound)?;

        if !dst.is_file() {
            return Err(FsError::InvalidArgument {
                reason: "destination must be a regular file",
            });
        }

        // Quantum entanglement: share the data
        dst.data_cache = src_data.0;
        dst.phys.size = src_data.1;
        dst.master_node = src_data.2;
        dst.dirty = true;

        // Increment reference count in dedup table if there's actual data
        use crate::dedup::dedup::DDT;
        if src_data.2.offset != 0 || src_data.2.vdev != 0 {
            let mut ddt = DDT.lock();
            if let Some(entry) = ddt.table.values_mut().find(|e| e.dva == src_data.2) {
                entry.ref_count += 1;
            }
        }

        Ok(())
    }

    /// Create a symbolic link.
    ///
    /// Creates a new symlink znode that points to the specified target path.
    /// The target is stored in the znode's data_cache.
    ///
    /// # Arguments
    ///
    /// * `dir_id` - Parent directory object ID
    /// * `name` - Name of the symlink
    /// * `target` - Target path (what the symlink points to)
    /// * `uid` - Owner user ID
    /// * `gid` - Owner group ID
    ///
    /// # Returns
    ///
    /// Object ID of the new symlink.
    pub fn symlink(
        &mut self,
        dir_id: u64,
        name: &str,
        target: &str,
        uid: u32,
        gid: u32,
    ) -> FsResult<u64> {
        // Check parent is a directory
        let dir = self.znodes.get(&dir_id).ok_or(FsError::NotFound)?;
        if !dir.is_dir() {
            return Err(FsError::NotDirectory);
        }

        // Create DMU transaction
        let mut tx = dmu_tx_create(&self.objset);
        let _ = dmu_tx_hold_write(&mut tx, dir_id, 0, 512);
        dmu_tx_assign(&mut tx, &self.objset, TxWaitType::Wait)?;

        // Allocate new object via DMU
        let object_id = self.alloc_object_id(&tx)?;

        // Create symlink znode
        let mut znode = Znode::new_symlink(object_id, dir_id, uid, gid);

        // Store target path in data_cache
        znode.data_cache = Some(target.as_bytes().to_vec());
        znode.phys.size = target.len() as u64;

        self.znodes.insert(object_id, znode);

        // Link into directory
        self.link_create(dir_id, name, object_id)?;

        // Commit transaction
        dmu_tx_commit(tx);

        Ok(object_id)
    }

    /// Read the target of a symbolic link.
    ///
    /// # Arguments
    ///
    /// * `object_id` - Object ID of the symlink
    ///
    /// # Returns
    ///
    /// The target path as a string.
    pub fn readlink(&self, object_id: u64) -> FsResult<String> {
        let znode = self.znodes.get(&object_id).ok_or(FsError::NotFound)?;

        if !znode.is_symlink() {
            return Err(FsError::InvalidArgument {
                reason: "not a symbolic link",
            });
        }

        // Get target from data_cache
        let target_bytes = znode.data_cache.as_ref().ok_or(FsError::InvalidArgument {
            reason: "symlink has no target",
        })?;

        String::from_utf8(target_bytes.clone()).map_err(|_| FsError::InvalidArgument {
            reason: "symlink target is not valid UTF-8",
        })
    }

    /// Read directory entries
    pub fn readdir(&self, dir_id: u64) -> FsResult<Vec<DirEntry>> {
        let dir = self.znodes.get(&dir_id).ok_or(FsError::NotFound)?;

        if !dir.is_dir() {
            return Err(FsError::NotDirectory);
        }

        let mut entries = Vec::new();

        // Always include . and ..
        entries.push(DirEntry {
            name: String::from("."),
            object_id: dir_id,
            file_type: 4, // DT_DIR
        });
        entries.push(DirEntry {
            name: String::from(".."),
            object_id: dir.phys.parent,
            file_type: 4, // DT_DIR
        });

        // Iterate directory entries (from in-memory ZAP)
        if let Some(dir_map) = self.dir_entries.get(&dir_id) {
            for (name, &object_id) in dir_map {
                let file_type = self
                    .znodes
                    .get(&object_id)
                    .map(|z| z.dirent_type())
                    .unwrap_or(0);

                entries.push(DirEntry {
                    name: name.clone(),
                    object_id,
                    file_type,
                });
            }
        }

        Ok(entries)
    }

    /// Sync file to disk (fsync)
    pub fn fsync(&mut self, object_id: u64) -> FsResult<()> {
        use crate::storage::zil::{ZilEngine, ZilOpcode};

        let znode = self.znodes.get_mut(&object_id).ok_or(FsError::NotFound)?;

        if znode.dirty {
            // Log to ZIL for crash consistency
            if let Some(data) = &znode.data_cache {
                let txg = self
                    .objset
                    .os_txg
                    .load(core::sync::atomic::Ordering::Acquire);
                let _ = ZilEngine::log_operation(txg, ZilOpcode::Write, object_id, 0, data);
            }

            // Flush ZIL to SLOG
            ZilEngine::flush_to_slog().map_err(|e| FsError::InvalidPoolConfig { reason: e })?;

            znode.dirty = false;
        }

        Ok(())
    }

    /// Sync the entire filesystem (commit TXG)
    ///
    /// Writes all dirty metadata and data to disk. This is called by Pool::sync().
    pub fn txg_sync(&mut self, dev_id: usize) -> FsResult<u64> {
        self.objset.txg_sync(dev_id)
    }

    // ============================================================================
    // SEND/RECEIVE OPERATIONS (from OpenZFS zfs_send.c / zfs_receive.c)
    // ============================================================================

    /// Serialize all objects to a send stream (full send)
    pub fn send_to_stream(&self, stream: &mut crate::net::send_recv::SendStream) -> FsResult<()> {
        use crate::fscore::structs::DnodePhys;

        crate::lcpfs_println!("[ SEND ] Serializing {} znodes", self.znodes.len());

        // Send all znodes (files and directories)
        for (&object_id, znode) in self.znodes.iter() {
            // Create a minimal dnode for the object
            let dnode = DnodePhys::zero();
            // Write object record (simplified - just send mode/size)
            stream.write_object(object_id, znode.phys.mode as u8, &dnode);

            // Write data blocks if file has cached data
            if znode.is_file() {
                if let Some(ref data) = znode.data_cache {
                    stream.write_block(object_id, 0, data);
                }
            }
        }

        // Send directory structure
        for (&dir_id, entries) in self.dir_entries.iter() {
            for (name, &child_id) in entries.iter() {
                // Encode directory entry as a small write
                let entry_data = format!("{}:{}", name, child_id);
                stream.write_block(dir_id, child_id, entry_data.as_bytes());
            }
        }

        crate::lcpfs_println!("[ SEND ] Stream complete: {} bytes", stream.size());
        Ok(())
    }

    /// Serialize only objects modified since from_txg (incremental send)
    pub fn send_incremental_to_stream(
        &self,
        stream: &mut crate::net::send_recv::SendStream,
        from_txg: u64,
    ) -> FsResult<()> {
        use crate::fscore::structs::DnodePhys;

        crate::lcpfs_println!("[ SEND ] Incremental send from TXG {}", from_txg);
        let mut sent_count = 0;

        // Send only dirty znodes or those modified after from_txg
        // In a full implementation, each znode would track its birth_txg and dirty_txg
        // For now, we use the dirty flag as a proxy
        for (&object_id, znode) in self.znodes.iter() {
            // Skip objects that haven't been modified
            if !znode.dirty {
                continue;
            }

            // Create a minimal dnode for the object
            let dnode = DnodePhys::zero();
            stream.write_object(object_id, znode.phys.mode as u8, &dnode);
            sent_count += 1;

            // Write data blocks if file has cached data
            if znode.is_file() {
                if let Some(ref data) = znode.data_cache {
                    stream.write_block(object_id, 0, data);
                }
            }
        }

        // Send only modified directory entries
        // In a full implementation, directory entries would also track TXG
        for (&dir_id, entries) in self.dir_entries.iter() {
            // Only send directory if the directory znode itself is dirty
            if let Some(dir_znode) = self.znodes.get(&dir_id) {
                if dir_znode.dirty {
                    for (name, &child_id) in entries.iter() {
                        let entry_data = format!("{}:{}", name, child_id);
                        stream.write_block(dir_id, child_id, entry_data.as_bytes());
                    }
                }
            }
        }

        crate::lcpfs_println!(
            "[ SEND ] Incremental send complete: {} objects, {} bytes",
            sent_count,
            stream.size()
        );
        Ok(())
    }

    /// Deserialize objects from a receive stream
    pub fn receive_from_stream(
        &mut self,
        recv: &mut crate::net::send_recv::ReceiveStream,
    ) -> FsResult<()> {
        crate::lcpfs_println!("[ RECV ] Processing stream...");

        let mut received_objects = 0;
        let mut received_blocks = 0;

        // Process all records in the stream
        while let Some((record, payload)) = recv
            .read_next_record()
            .map_err(|e| FsError::InvalidPoolConfig { reason: e })?
        {
            match record.record_type {
                3 => {
                    // SendRecordType::Object
                    // Create new znode from dnode metadata
                    let object_id = record.object_id;

                    // Create basic znode (would parse dnode from payload in real impl)
                    let znode = Znode::new_file(object_id, 0, 0o644, 0, 0);
                    self.znodes.insert(object_id, znode);

                    received_objects += 1;
                }
                5 => {
                    // SendRecordType::Write
                    // Write data block to object
                    let object_id = record.object_id;

                    if let Some(znode) = self.znodes.get_mut(&object_id) {
                        let payload_len = payload.len() as u64;
                        znode.data_cache = Some(payload);
                        znode.phys.size = payload_len;
                    }

                    received_blocks += 1;
                }
                _ => {
                    // Skip unknown record types
                }
            }
        }

        crate::lcpfs_println!(
            "[ RECV ] Received {} objects, {} blocks",
            received_objects,
            received_blocks
        );
        Ok(())
    }
}

// ================================================================================
// GLOBAL ZPL INSTANCE
// ================================================================================

use lazy_static::lazy_static;

lazy_static! {
    /// Global ZPL filesystem instance
    pub static ref ZPL: Mutex<Zpl> = Mutex::new(Zpl::new());
}

// ================================================================================
// HELPER FUNCTIONS
// ================================================================================

/// Get current time as [seconds, nanoseconds]
fn get_current_time() -> [u64; 2] {
    // Use the unified time provider for consistent timestamps
    let secs = crate::time::now();
    let ns = crate::time::high_resolution() % 1_000_000_000; // Use TSC for sub-second precision
    [secs, ns]
}

// ================================================================================
// CONVENIENCE FUNCTIONS (PUBLIC API)
// ================================================================================

/// Create a file at the given path
pub fn zpl_create(path: &str, mode: u32) -> FsResult<u64> {
    let mut zpl = ZPL.lock();

    // Parse path to find parent directory and filename
    let (parent_id, name) = parse_path(&zpl, path)?;

    zpl.create(parent_id, name, mode, 0, 0)
}

/// Create a directory at the given path
pub fn zpl_mkdir(path: &str, mode: u32) -> FsResult<u64> {
    let mut zpl = ZPL.lock();

    let (parent_id, name) = parse_path(&zpl, path)?;

    zpl.mkdir(parent_id, name, mode, 0, 0)
}

/// Open a file and return handle
pub fn zpl_open(path: &str, flags: u32) -> FsResult<u64> {
    let mut zpl = ZPL.lock();

    let object_id = resolve_path(&zpl, path)?;

    zpl.open(object_id, flags)
}

/// Close a file handle
pub fn zpl_close(handle: u64) -> FsResult<()> {
    ZPL.lock().close(handle)
}

/// Read from file handle
pub fn zpl_read(handle: u64, buf: &mut [u8]) -> FsResult<usize> {
    ZPL.lock().read(handle, buf)
}

/// Write to file handle
pub fn zpl_write(handle: u64, buf: &[u8]) -> FsResult<usize> {
    ZPL.lock().write(handle, buf)
}

/// Get file stats
pub fn zpl_stat(path: &str) -> FsResult<FileStat> {
    let zpl = ZPL.lock();
    let object_id = resolve_path(&zpl, path)?;
    zpl.getattr(object_id)
}

/// Read directory contents
pub fn zpl_readdir(path: &str) -> FsResult<Vec<DirEntry>> {
    let zpl = ZPL.lock();
    let object_id = resolve_path(&zpl, path)?;
    zpl.readdir(object_id)
}

/// Unlink (delete) a file at the given path
pub fn zpl_unlink(path: &str) -> FsResult<()> {
    let mut zpl = ZPL.lock();
    let (parent_id, name) = parse_path(&zpl, path)?;
    zpl.unlink(parent_id, name)
}

/// Remove a directory at the given path
pub fn zpl_rmdir(path: &str) -> FsResult<()> {
    let mut zpl = ZPL.lock();
    let (parent_id, name) = parse_path(&zpl, path)?;
    zpl.rmdir(parent_id, name)
}

/// Rename a file or directory
///
/// Atomically moves `src_path` to `dst_path`. If `dst_path` exists:
/// - For files: overwrites the destination
/// - For directories: destination must be empty
pub fn zpl_rename(src_path: &str, dst_path: &str) -> FsResult<()> {
    let mut zpl = ZPL.lock();
    let (src_parent_id, src_name) = parse_path(&zpl, src_path)?;
    let (dst_parent_id, dst_name) = parse_path(&zpl, dst_path)?;
    zpl.rename(src_parent_id, src_name, dst_parent_id, dst_name)
}

// ================================================================================
// PATH RESOLUTION
// ================================================================================

/// Parse a path into (parent_object_id, filename)
fn parse_path<'a>(zpl: &Zpl, path: &'a str) -> FsResult<(u64, &'a str)> {
    let path = path.trim_start_matches('/');

    if path.is_empty() {
        return Err(FsError::InvalidArgument {
            reason: "empty path",
        });
    }

    let mut current = zpl.root_id;
    let parts: Vec<&str> = path.split('/').collect();

    // Navigate to parent directory
    for part in parts.iter().take(parts.len().saturating_sub(1)) {
        if part.is_empty() {
            continue;
        }

        // Look up in current directory
        // In real impl, use ZAP lookup
        // For now, fail since we don't have name mapping
        return Err(FsError::NotFound);
    }

    let filename = parts.last().ok_or(FsError::InvalidArgument {
        reason: "no filename",
    })?;
    Ok((current, filename))
}

/// Resolve a full path to an object ID
fn resolve_path(zpl: &Zpl, path: &str) -> FsResult<u64> {
    let path = path.trim_start_matches('/');

    if path.is_empty() {
        return Ok(zpl.root_id);
    }

    let mut current = zpl.root_id;

    for part in path.split('/') {
        if part.is_empty() {
            continue;
        }

        current = zpl.lookup(current, part)?;
    }

    Ok(current)
}

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

    #[test]
    fn test_flock_acquire_exclusive() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        // Process 1 acquires exclusive lock
        assert!(znode.flock_acquire(LockType::Exclusive, 1).is_ok());

        // Process 2 cannot acquire any lock
        assert_eq!(
            znode.flock_acquire(LockType::Exclusive, 2),
            Err(FsError::ResourceBusy)
        );
        assert_eq!(
            znode.flock_acquire(LockType::Shared, 2),
            Err(FsError::ResourceBusy)
        );
    }

    #[test]
    fn test_flock_acquire_shared() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        // Process 1 acquires shared lock
        assert!(znode.flock_acquire(LockType::Shared, 1).is_ok());

        // Process 2 can also acquire shared lock
        assert!(znode.flock_acquire(LockType::Shared, 2).is_ok());

        // But cannot acquire exclusive lock
        assert_eq!(
            znode.flock_acquire(LockType::Exclusive, 3),
            Err(FsError::ResourceBusy)
        );
    }

    #[test]
    fn test_flock_release() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        znode
            .flock_acquire(LockType::Exclusive, 1)
            .expect("test: operation should succeed");
        assert!(znode.flock_release(1).is_ok());

        // After release, another process can lock
        assert!(znode.flock_acquire(LockType::Exclusive, 2).is_ok());
    }

    #[test]
    fn test_fcntl_lock_exclusive() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        let lock1 = FileLock {
            lock_type: LockType::Exclusive,
            start: 0,
            length: 100,
            pid: 1,
        };

        assert!(znode.fcntl_lock(lock1.clone()).is_ok());

        // Overlapping lock from different process should fail
        let lock2 = FileLock {
            lock_type: LockType::Exclusive,
            start: 50,
            length: 100,
            pid: 2,
        };

        assert_eq!(znode.fcntl_lock(lock2), Err(FsError::ResourceBusy));
    }

    #[test]
    fn test_fcntl_lock_non_overlapping() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        let lock1 = FileLock {
            lock_type: LockType::Exclusive,
            start: 0,
            length: 100,
            pid: 1,
        };

        let lock2 = FileLock {
            lock_type: LockType::Exclusive,
            start: 200,
            length: 100,
            pid: 2,
        };

        assert!(znode.fcntl_lock(lock1).is_ok());
        assert!(znode.fcntl_lock(lock2).is_ok());
    }

    #[test]
    fn test_fcntl_lock_shared() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        let lock1 = FileLock {
            lock_type: LockType::Shared,
            start: 0,
            length: 100,
            pid: 1,
        };

        let lock2 = FileLock {
            lock_type: LockType::Shared,
            start: 50,
            length: 100,
            pid: 2,
        };

        // Shared locks can overlap
        assert!(znode.fcntl_lock(lock1).is_ok());
        assert!(znode.fcntl_lock(lock2).is_ok());
    }

    #[test]
    fn test_fcntl_unlock() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        let lock = FileLock {
            lock_type: LockType::Exclusive,
            start: 0,
            length: 100,
            pid: 1,
        };

        znode
            .fcntl_lock(lock)
            .expect("test: operation should succeed");
        assert_eq!(znode.record_locks.len(), 1);

        znode
            .fcntl_unlock(0, 100, 1)
            .expect("test: operation should succeed");
        assert_eq!(znode.record_locks.len(), 0);
    }

    #[test]
    fn test_fcntl_test_lock() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        let lock1 = FileLock {
            lock_type: LockType::Exclusive,
            start: 0,
            length: 100,
            pid: 1,
        };

        znode
            .fcntl_lock(lock1.clone())
            .expect("test: operation should succeed");

        // Test lock from different process should detect conflict
        let test_lock = FileLock {
            lock_type: LockType::Exclusive,
            start: 50,
            length: 100,
            pid: 2,
        };

        let conflict = znode.fcntl_test_lock(&test_lock);
        assert!(conflict.is_some());
        assert_eq!(conflict.expect("test: operation should succeed").pid, 1);
    }

    #[test]
    fn test_release_all_locks() {
        let mut znode = Znode::new_file(1, 0, 0o644, 1000, 1000);

        // Add both flock and fcntl locks
        znode
            .flock_acquire(LockType::Exclusive, 1)
            .expect("test: operation should succeed");

        let lock = FileLock {
            lock_type: LockType::Exclusive,
            start: 0,
            length: 100,
            pid: 1,
        };
        znode
            .fcntl_lock(lock)
            .expect("test: operation should succeed");

        assert!(znode.flock.is_some());
        assert_eq!(znode.record_locks.len(), 1);

        // Release all locks for process 1
        znode.release_all_locks(1);

        assert!(znode.flock.is_none());
        assert_eq!(znode.record_locks.len(), 0);
    }

    #[test]
    fn test_file_lock_conflicts() {
        let lock1 = FileLock {
            lock_type: LockType::Exclusive,
            start: 0,
            length: 100,
            pid: 1,
        };

        let lock2 = FileLock {
            lock_type: LockType::Exclusive,
            start: 50,
            length: 100,
            pid: 2,
        };

        let lock3 = FileLock {
            lock_type: LockType::Shared,
            start: 0,
            length: 100,
            pid: 1,
        };

        let lock4 = FileLock {
            lock_type: LockType::Shared,
            start: 50,
            length: 100,
            pid: 2,
        };

        // Exclusive locks conflict
        assert!(lock1.conflicts_with(&lock2));

        // Exclusive and shared conflict
        assert!(lock1.conflicts_with(&lock4));

        // Shared locks don't conflict
        assert!(!lock3.conflicts_with(&lock4));
    }
}