use crate::vfs::open_flags::OpenFlags;
use crate::vfs::{VfsBackend, VfsError, VfsFile, VfsStat};
use crate::ssh_server::upload_hook::UploadHook;
use crate::webdav_version::WebDavVersioning;
use bytes::{Buf, Bytes};
use dav_server::davpath::DavPath;
use dav_server::fs::{
    DavDirEntry, DavFile, DavFileSystem, DavMetaData, DavProp, FsError, FsFuture, FsStream,
    OpenOptions,
};
use dav_server::ls::DavLockSystem;
use http::StatusCode;
use dav_server::memls::MemLs;
use futures_util::stream;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::{Arc, Mutex, RwLock};
use std::time::SystemTime;

use crate::webdav_locks::PersistedLs;

fn map_vfs_error(e: VfsError) -> FsError {
    match e {
        VfsError::NotFound(_) => FsError::NotFound,
        VfsError::PermissionDenied(_) => FsError::Forbidden,
        VfsError::Unsupported(_) => FsError::NotImplemented,
        VfsError::AlreadyExists(_) => FsError::Exists,
        VfsError::NotEmpty(_) => FsError::Forbidden,
        VfsError::NotADirectory(_) => FsError::Forbidden,
        VfsError::IsADirectory(_) => FsError::Forbidden,
        VfsError::Io(_) | VfsError::UnexpectedEof => FsError::GeneralFailure,
    }
}

/// Expected credentials for WebDAV Basic Auth validation
#[derive(Clone)]
#[derive(Clone)]
pub struct WebdavCredentials {
    pub username: String,
    pub password: Option<String>,
}

/// Serializable dead property for disk persistence
#[derive(Serialize, Deserialize)]
struct DeadPropEntry {
    name: String,
    prefix: Option<String>,
    namespace: Option<String>,
    xml: Option<Vec<u8>>,
}

impl From<&DavProp> for DeadPropEntry {
    fn from(p: &DavProp) -> Self {
        Self {
            name: p.name.clone(),
            prefix: p.prefix.clone(),
            namespace: p.namespace.clone(),
            xml: p.xml.clone(),
        }
    }
}

impl From<DeadPropEntry> for DavProp {
    fn from(e: DeadPropEntry) -> Self {
        Self {
            name: e.name,
            prefix: e.prefix,
            namespace: e.namespace,
            xml: e.xml,
        }
    }
}

pub struct VfsDavFs {
    vfs: Box<dyn VfsBackend>,
    root: PathBuf,
    upload_hook: Option<Arc<UploadHook>>,
    user_uuid: String,
    versioning: Option<Arc<WebDavVersioning>>,
    props_data: Arc<RwLock<HashMap<String, Vec<DavProp>>>>,
    props_path: PathBuf,
    enable_acl: bool,
}

impl Clone for VfsDavFs {
    fn clone(&self) -> Self {
        Self {
            vfs: self.vfs.clone_boxed(),
            root: self.root.clone(),
            upload_hook: self.upload_hook.clone(),
            user_uuid: self.user_uuid.clone(),
            versioning: self.versioning.clone(),
            props_data: self.props_data.clone(),
            props_path: self.props_path.clone(),
            enable_acl: self.enable_acl,
        }
    }
}

impl VfsDavFs {
    fn dead_props_path(root: &Path) -> PathBuf {
        root.join(".webdav_props.json")
    }

    fn load_props(vfs: &dyn VfsBackend, root: &Path) -> HashMap<String, Vec<DavProp>> {
        let path = Self::dead_props_path(root);
        let flags = OpenFlags::new().read();
        let json = match vfs.open_file(&path, &flags) {
            Ok(mut file) => {
                let mut buf = Vec::new();
                let mut chunk = [0u8; 4096];
                loop {
                    match file.read(&mut chunk) {
                        Ok(0) => break,
                        Ok(n) => buf.extend_from_slice(&chunk[..n]),
                        Err(_) => return HashMap::new(),
                    }
                }
                String::from_utf8_lossy(&buf).to_string()
            }
            Err(_) => return HashMap::new(),
        };
        let entries: HashMap<String, Vec<DeadPropEntry>> = match serde_json::from_str(&json) {
            Ok(m) => m,
            Err(_) => HashMap::new(),
        };
        entries
            .into_iter()
            .map(|(k, v)| (k, v.into_iter().map(DavProp::from).collect()))
            .collect()
    }

    fn save_props(&self) {
        let data = match self.props_data.read() {
            Ok(guard) => guard,
            Err(e) => {
                log::warn!("props_data RwLock poisoned in save_props, recovering");
                e.into_inner()
            }
        };
        let entries: HashMap<String, Vec<DeadPropEntry>> = data
            .iter()
            .filter(|(_, v)| !v.is_empty())
            .map(|(k, v)| (k.clone(), v.iter().map(DeadPropEntry::from).collect()))
            .collect();
        if let Ok(json) = serde_json::to_string(&entries) {
            let path = self.root.join(".webdav_props.json");
            let flags = OpenFlags::new().write().create().truncate().mode(0o644);
            match self.vfs.open_file(&path, &flags) {
                Ok(mut file) => {
                    if let Err(e) = file.write_all(json.as_bytes()) {
                        log::warn!("save_props write_all failed: {:?}", e);
                    }
                    if let Err(e) = file.flush() {
                        log::warn!("save_props flush failed: {:?}", e);
                    }
                }
                Err(e) => {
                    log::warn!("save_props open_file failed: {:?}", e);
                }
            }
        }
    }

    pub fn new(
        vfs: Box<dyn VfsBackend>,
        root: PathBuf,
        upload_hook: Option<Arc<UploadHook>>,
        user_uuid: String,
    ) -> Box<Self> {
        let props_path = Self::dead_props_path(&root);
        let props_data = Arc::new(RwLock::new(Self::load_props(vfs.as_ref(), &root)));
        Box::new(Self {
            vfs,
            root,
            upload_hook,
            user_uuid,
            versioning: None,
            props_data,
            props_path,
            enable_acl: true,
        })
    }

    pub fn with_versioning(
        vfs: Box<dyn VfsBackend>,
        root: PathBuf,
        upload_hook: Option<Arc<UploadHook>>,
        user_uuid: String,
        versioning: Arc<WebDavVersioning>,
    ) -> Box<Self> {
        let props_path = Self::dead_props_path(&root);
        let props_data = Arc::new(RwLock::new(Self::load_props(vfs.as_ref(), &root)));
        Box::new(Self {
            vfs,
            root,
            upload_hook,
            user_uuid,
            versioning: Some(versioning),
            props_data,
            props_path,
            enable_acl: true,
        })
    }

    pub fn set_enable_acl(&mut self, enable: bool) {
        self.enable_acl = enable;
    }

    fn rel_key(&self, path: &DavPath) -> String {
        let rel = path.as_pathbuf();
        rel.to_string_lossy().to_string()
    }

    fn resolve_path(&self, path: &DavPath) -> Result<PathBuf, FsError> {
        let relative = path.as_rel_ospath();
        let full = self.root.join(relative);

        // Path traversal protection: canonicalize root once
        let root_canonical = self.root.canonicalize().map_err(|_| FsError::NotFound)?;

        // If path exists, use canonicalized version
        if let Ok(canonical) = full.canonicalize() {
            if !canonical.starts_with(&root_canonical) {
                return Err(FsError::NotFound);
            }
            return Ok(canonical);
        }

        // Path doesn't exist yet (e.g., new file/dir being created).
        // Validate parent directory is within root and no .. traversal.
        let parent = full.parent().ok_or(FsError::NotFound)?;
        let parent_canonical = parent.canonicalize().map_err(|_| FsError::NotFound)?;
        if !parent_canonical.starts_with(&root_canonical) {
            return Err(FsError::NotFound);
        }

        // Sanity check: ensure relative path doesn't contain ".."
        if relative.components().any(|c| c == std::path::Component::ParentDir) {
            return Err(FsError::NotFound);
        }

        Ok(full)
    }
}

#[derive(Debug, Clone)]
pub struct VfsDavMetaData {
    len: u64,
    is_dir: bool,
    modified: SystemTime,
    accessed: SystemTime,
    mode: u32,
}

impl VfsDavMetaData {
    pub fn new(len: u64, is_dir: bool) -> Self {
        let now = SystemTime::now();
        Self {
            len,
            is_dir,
            modified: now,
            accessed: now,
            mode: 0o644,
        }
    }

    pub fn from_stat(stat: &VfsStat) -> Self {
        Self {
            len: stat.size,
            is_dir: stat.is_dir,
            modified: stat.mtime,
            accessed: stat.atime,
            mode: stat.mode,
        }
    }
}

impl DavMetaData for VfsDavMetaData {
    fn len(&self) -> u64 {
        self.len
    }

    fn modified(&self) -> Result<SystemTime, FsError> {
        Ok(self.modified)
    }

    fn is_dir(&self) -> bool {
        self.is_dir
    }

    fn accessed(&self) -> Result<SystemTime, FsError> {
        Ok(self.accessed)
    }

    fn executable(&self) -> Result<bool, FsError> {
        Ok(!self.is_dir && (self.mode & 0o111 != 0))
    }
}

#[derive(Debug, Clone)]
pub struct VfsDavDirEntry {
    name: String,
    meta: VfsDavMetaData,
}

impl VfsDavDirEntry {
    pub fn new(name: String, meta: VfsDavMetaData) -> Self {
        Self { name, meta }
    }
}

impl DavDirEntry for VfsDavDirEntry {
    fn name(&self) -> Vec<u8> {
        self.name.as_bytes().to_vec()
    }

    fn metadata(&self) -> FsFuture<'_, Box<dyn DavMetaData>> {
        Box::pin(std::future::ready(Ok(Box::new(self.meta.clone()) as Box<dyn DavMetaData>)))
    }
}

pub struct VfsDavFile {
    data: Vec<u8>,
    position: u64,
    vfs_file: Option<Mutex<Box<dyn VfsFile>>>,
    vfs: Option<Box<dyn VfsBackend>>,
    path: Option<PathBuf>,
    upload_hook: Option<Arc<UploadHook>>,
    user_uuid: String,
    is_write: bool,
    versioning: Option<Arc<WebDavVersioning>>,
    flushed: bool,
    read_cache: Vec<u8>,
    read_cache_offset: u64,
}

impl std::fmt::Debug for VfsDavFile {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("VfsDavFile")
            .field("is_write", &self.is_write)
            .field("path", &self.path)
            .field("data_len", &self.data.len())
            .field("position", &self.position)
            .finish()
    }
}

impl VfsDavFile {
    pub fn new_read(vfs_file: Box<dyn VfsFile>) -> Self {
        Self {
            data: Vec::new(),
            position: 0,
            vfs_file: Some(std::sync::Mutex::new(vfs_file)),
            vfs: None,
            path: None,
            upload_hook: None,
            user_uuid: String::new(),
            is_write: false,
            versioning: None,
            flushed: true,
            read_cache: Vec::new(),
            read_cache_offset: 0,
        }
    }

    pub fn new_write(
        path: PathBuf,
        vfs: Box<dyn VfsBackend>,
        upload_hook: Option<Arc<UploadHook>>,
        user_uuid: String,
        versioning: Option<Arc<WebDavVersioning>>,
    ) -> Self {
        if let Some(parent) = path.parent() {
            let _ = vfs.create_dir_all(parent, 0o755);
        }
        let flags = OpenFlags::new().write().create().truncate().mode(0o644);
        let vfs_file = vfs.open_file(&path, &flags).ok()
            .map(std::sync::Mutex::new);
        Self {
            data: Vec::new(),
            position: 0,
            vfs_file,
            vfs: Some(vfs),
            path: Some(path),
            upload_hook,
            user_uuid,
            is_write: true,
            versioning,
            flushed: false,
            read_cache: Vec::new(),
            read_cache_offset: 0,
        }
    }
}

impl Drop for VfsDavFile {
    fn drop(&mut self) {
        if self.is_write && !self.flushed && !self.data.is_empty() {
            if let Some(path) = &self.path {
                log::error!(
                    "VfsDavFile dropped with {} bytes of unwritten data for {:?} - DATA LOSS!",
                    self.data.len(),
                    path
                );
            }
        }
    }
}

impl DavFile for VfsDavFile {
    fn metadata(&'_ mut self) -> FsFuture<'_, Box<dyn DavMetaData>> {
        if let Some(vfs_file_mutex) = &self.vfs_file {
            if let Ok(mut vfs_file) = vfs_file_mutex.lock() {
                if let Ok(stat) = vfs_file.stat() {
                    return Box::pin(std::future::ready(Ok(
                        Box::new(VfsDavMetaData::from_stat(&stat)) as Box<dyn DavMetaData>,
                    )));
                }
            }
        }
        let len = self.data.len() as u64;
        Box::pin(std::future::ready(Ok(
            Box::new(VfsDavMetaData::new(len, false)) as Box<dyn DavMetaData>,
        )))
    }

    fn write_buf(&'_ mut self, mut buf: Box<dyn Buf + Send>) -> FsFuture<'_, ()> {
        if let Some(vfs_file_mutex) = &self.vfs_file {
            if let Ok(mut vfs_file) = vfs_file_mutex.lock() {
                while buf.has_remaining() {
                    let chunk = buf.chunk();
                    let _ = vfs_file.write_all(chunk);
                    self.data.extend_from_slice(chunk);
                    buf.advance(chunk.len());
                }
                return Box::pin(std::future::ready(Ok(())));
            }
        }
        while buf.has_remaining() {
            let chunk = buf.chunk();
            self.data.extend_from_slice(chunk);
            buf.advance(chunk.len());
        }
        Box::pin(std::future::ready(Ok(())))
    }

    fn write_bytes(&'_ mut self, buf: Bytes) -> FsFuture<'_, ()> {
        if let Some(vfs_file_mutex) = &self.vfs_file {
            if let Ok(mut vfs_file) = vfs_file_mutex.lock() {
                let _ = vfs_file.write_all(&buf);
            }
        }
        self.data.extend_from_slice(&buf);
        Box::pin(std::future::ready(Ok(())))
    }

    fn read_bytes(&'_ mut self, count: usize) -> FsFuture<'_, Bytes> {
        const CHUNK_SIZE: usize = 64 * 1024; // 64KB read-ahead

        if let Some(vfs_file_mutex) = &self.vfs_file {
            // Check if requested data is in cache
            let cache_start = self.read_cache_offset as usize;
            let cache_end = cache_start + self.read_cache.len();
            let req_start = self.position as usize;
            let req_end = req_start + count;

            if req_start >= cache_start && req_end <= cache_end {
                // Data in cache - return directly
                let offset_in_cache = req_start - cache_start;
                let bytes = Bytes::copy_from_slice(
                    &self.read_cache[offset_in_cache..offset_in_cache + count.min(self.read_cache.len() - offset_in_cache)]
                );
                self.position += bytes.len() as u64;
                return Box::pin(std::future::ready(Ok(bytes)));
            }

            // Need to read new chunk
            if let Ok(mut vfs_file) = vfs_file_mutex.lock() {
                // Seek to current position if needed
                let current_pos = vfs_file.seek(std::io::SeekFrom::Current(0)).unwrap_or(self.position);
                if current_pos != self.position
                    && vfs_file.seek(std::io::SeekFrom::Start(self.position)).is_err() {
                        return Box::pin(std::future::ready(Err(FsError::GeneralFailure)));
                    }

                // Read larger chunk
                let read_size = std::cmp::max(count, CHUNK_SIZE);
                self.read_cache.resize(read_size, 0);
                self.read_cache_offset = self.position;

                match vfs_file.read(&mut self.read_cache) {
                    Ok(0) => {
                        self.read_cache.clear();
                        return Box::pin(std::future::ready(Ok(Bytes::new())));
                    }
                    Ok(n) => {
                        self.read_cache.truncate(n);
                        // Return requested portion
                        let result_len = count.min(n);
                        let bytes = Bytes::copy_from_slice(&self.read_cache[..result_len]);
                        self.position += result_len as u64;
                        return Box::pin(std::future::ready(Ok(bytes)));
                    }
                    Err(_) => {}
                }
            }
            Box::pin(std::future::ready(Err(FsError::NotImplemented)))
        } else {
            // Write mode - read from self.data buffer
            let start = self.position as usize;
            let end = std::cmp::min(start + count, self.data.len());

            if start >= self.data.len() {
                Box::pin(std::future::ready(Ok(Bytes::new())))
            } else {
                let bytes = Bytes::copy_from_slice(&self.data[start..end]);
                self.position = end as u64;
                Box::pin(std::future::ready(Ok(bytes)))
            }
        }
    }

    fn seek(&'_ mut self, pos: std::io::SeekFrom) -> FsFuture<'_, u64> {
        if let Some(vfs_file_mutex) = &self.vfs_file {
            if let Ok(mut vfs_file) = vfs_file_mutex.lock() {
                if let Ok(new_pos) = vfs_file.seek(pos) {
                    self.position = new_pos;
                    self.read_cache.clear(); // Invalidate cache on seek
                    self.read_cache_offset = 0;
                    return Box::pin(std::future::ready(Ok(new_pos)));
                }
            }
            Box::pin(std::future::ready(Err(FsError::NotImplemented)))
        } else {
            let new_pos = match pos {
                std::io::SeekFrom::Start(offset) => offset,
                std::io::SeekFrom::Current(offset) => {
                    (self.position as i64 + offset) as u64
                }
                std::io::SeekFrom::End(offset) => {
                    (self.data.len() as i64 + offset) as u64
                }
            };
            self.position = new_pos;
            Box::pin(std::future::ready(Ok(new_pos)))
        }
    }

    fn flush(&'_ mut self) -> FsFuture<'_, ()> {
        if !self.is_write {
            return Box::pin(std::future::ready(Ok(())));
        }
        if self.flushed {
            return Box::pin(std::future::ready(Ok(())));
        }

        // Quota check before write
        if !self.data.is_empty() {
            if let Some(vfs) = &self.vfs {
                if let Some(path) = &self.path {
                    let usage = vfs.get_quota_usage(path);
                    let quota = vfs.get_quota(path);
                    if let (Ok(usage), Ok(quota)) = (usage, quota) {
                        if quota.space_limit > 0 {
                            let new_size = usage.space_used + self.data.len() as u64;
                            if new_size > quota.space_limit {
                                log::warn!(
                                    "Quota exceeded: current={}, adding={}, limit={}",
                                    usage.space_used,
                                    self.data.len(),
                                    quota.space_limit
                                );
                                return Box::pin(std::future::ready(Err(FsError::InsufficientStorage)));
                            }
                        }
                    }
                }
            }
        }

        // Phase 1: Flush to storage
        if let Some(vfs_file_mutex) = &self.vfs_file {
            match vfs_file_mutex.lock() {
                Ok(mut vfs_file) => {
                    if let Err(e) = vfs_file.flush() {
                        log::error!("VfsDavFile::flush() VFS flush failed: {:?}", e);
                        return Box::pin(std::future::ready(Err(FsError::GeneralFailure)));
                    }
                }
                Err(e) => {
                    log::error!("VfsDavFile::flush() mutex poisoned: {:?}", e);
                    return Box::pin(std::future::ready(Err(FsError::GeneralFailure)));
                }
            }
        } else if !self.data.is_empty() {
            if let Some(path) = &self.path {
                if let Some(vfs) = &mut self.vfs {
                    let flags = OpenFlags::new().write().create().truncate().mode(0o644);
                    match vfs.open_file(path, &flags) {
                        Ok(mut vfs_file) => {
                            if let Err(e) = vfs_file.write_all(&self.data) {
                                log::error!("VfsDavFile::flush() write_all failed for {:?}: {:?}", path, e);
                                return Box::pin(std::future::ready(Err(FsError::GeneralFailure)));
                            }
                            if let Err(e) = vfs_file.flush() {
                                log::error!("VfsDavFile::flush() VFS flush failed for {:?}: {:?}", path, e);
                                return Box::pin(std::future::ready(Err(FsError::GeneralFailure)));
                            }
                        }
                        Err(e) => {
                            log::error!("VfsDavFile::flush() open_file failed for {:?}: {:?}", path, e);
                            return Box::pin(std::future::ready(Err(FsError::GeneralFailure)));
                        }
                    }
                }
            }
        }

        self.flushed = true;

        // Phase 2: Create version from buffered data (no disk re-read)
        if let (Some(versioning), Some(path)) = (&self.versioning, &self.path) {
            if !self.data.is_empty() {
                let path_str = path.to_string_lossy().to_string();
                if let Err(e) = versioning.create_version(&path_str, &self.data, None, None) {
                    log::warn!("VfsDavFile::flush() create_version failed for {:?}: {:?}", path, e);
                }
            }
        }

        // Phase 3: Clear buffered data
        self.data.clear();

        // Phase 4: Upload hook
        if let Some(path) = &self.path {
            if let Some(hook) = &self.upload_hook {
                if let Err(e) = hook.trigger(path, &self.user_uuid) {
                    log::warn!("Upload hook failed for {:?}: {}", path, e);
                }
            }
        }

        Box::pin(std::future::ready(Ok(())))
    }
}

impl VfsDavFs {
    fn empty_acl_xml() -> Vec<u8> {
        b"<D:acl xmlns:D=\"DAV:\"><D:ace><D:principal><D:all/></D:principal><D:grant><D:privilege><D:all/></D:privilege></D:grant></D:ace></D:acl>".to_vec()
    }

    fn is_acl_prop(prop: &DavProp) -> bool {
        prop.name == "acl" && prop.namespace.as_deref() == Some("DAV:")
    }

    fn acl_prop() -> DavProp {
        DavProp {
            name: "acl".to_string(),
            prefix: Some("D".to_string()),
            namespace: Some("DAV:".to_string()),
            xml: Some(Self::empty_acl_xml()),
        }
    }

    fn check_acl(&self, path: &Path, mask: crate::vfs::VfsAceMask) -> Result<(), FsError> {
        if !self.enable_acl {
            return Ok(());
        }
        match self.vfs.check_acl(path, &self.user_uuid, mask) {
            Ok(true) => Ok(()),
            Ok(false) => Err(FsError::Forbidden),
            Err(crate::vfs::VfsError::Unsupported(_)) => Ok(()),
            Err(_) => Err(FsError::Forbidden),
        }
    }
}

impl DavFileSystem for VfsDavFs {
    fn open<'a>(
        &'a self,
        path: &'a DavPath,
        options: OpenOptions,
    ) -> FsFuture<'a, Box<dyn DavFile>> {
        let full_path = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };

        if options.write {
            if let Err(e) = self.check_acl(&full_path, crate::vfs::VfsAceMask::WriteData) {
                return Box::pin(std::future::ready(Err(e)));
            }
            let file = VfsDavFile::new_write(
                full_path,
                self.vfs.clone_boxed(),
                self.upload_hook.clone(),
                self.user_uuid.clone(),
                self.versioning.clone(),
            );
            Box::pin(std::future::ready(Ok(Box::new(file) as Box<dyn DavFile>)))
        } else {
            if let Err(e) = self.check_acl(&full_path, crate::vfs::VfsAceMask::ReadData) {
                return Box::pin(std::future::ready(Err(e)));
            }
            let flags = OpenFlags::new().read();
            match self.vfs.open_file(&full_path, &flags) {
                Ok(vfs_file) => {
                    let file = VfsDavFile::new_read(vfs_file);
                    Box::pin(std::future::ready(Ok(Box::new(file) as Box<dyn DavFile>)))
                }
                Err(_) => Box::pin(std::future::ready(Err(FsError::NotFound))),
            }
        }
    }

    fn read_dir<'a>(
        &'a self,
        path: &'a DavPath,
        _meta: dav_server::fs::ReadDirMeta,
    ) -> FsFuture<'a, FsStream<Box<dyn DavDirEntry>>> {
        let full_path = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };

        if let Err(e) = self.check_acl(&full_path, crate::vfs::VfsAceMask::ListDirectory) {
            return Box::pin(std::future::ready(Err(e)));
        }

        match self.vfs.read_dir(&full_path) {
            Ok(entries) => {
                let results: Vec<Box<dyn DavDirEntry>> = entries
                    .into_iter()
                    .map(|e| {
                        let meta = VfsDavMetaData::from_stat(&e.stat);
                        Box::new(VfsDavDirEntry::new(e.name, meta)) as Box<dyn DavDirEntry>
                    })
                    .collect();

                let stream = stream::iter(results.into_iter().map(Ok));
                Box::pin(std::future::ready(Ok(Box::pin(stream) as FsStream<Box<dyn DavDirEntry>>)))
            }
            Err(_) => Box::pin(std::future::ready(Err(FsError::NotFound))),
        }
    }

    fn metadata<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, Box<dyn DavMetaData>> {
        let full_path = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };

        match self.vfs.stat(&full_path) {
            Ok(stat) => {
                let meta = VfsDavMetaData::from_stat(&stat);
                Box::pin(std::future::ready(Ok(Box::new(meta) as Box<dyn DavMetaData>)))
            }
            Err(_) => Box::pin(std::future::ready(Err(FsError::NotFound))),
        }
    }

    fn create_dir<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, ()> {
        let full_path = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        if let Err(e) = self.check_acl(&full_path, crate::vfs::VfsAceMask::AddSubdirectory) {
            return Box::pin(std::future::ready(Err(e)));
        }
        if self.vfs.exists(&full_path) {
            return Box::pin(std::future::ready(Err(FsError::Exists)));
        }
        match self.vfs.create_dir_all(&full_path, 0o755) {
            Ok(_) => Box::pin(std::future::ready(Ok(()))),
            Err(e) => Box::pin(std::future::ready(Err(map_vfs_error(e)))),
        }
    }

    fn remove_dir<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, ()> {
        let full_path = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        if let Err(e) = self.check_acl(&full_path, crate::vfs::VfsAceMask::DeleteChild) {
            return Box::pin(std::future::ready(Err(e)));
        }
        match self.vfs.remove_dir_all(&full_path) {
            Ok(_) => Box::pin(std::future::ready(Ok(()))),
            Err(e) => Box::pin(std::future::ready(Err(map_vfs_error(e)))),
        }
    }

    fn remove_file<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, ()> {
        let full_path = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        if let Err(e) = self.check_acl(&full_path, crate::vfs::VfsAceMask::Delete) {
            return Box::pin(std::future::ready(Err(e)));
        }
        match self.vfs.remove_file(&full_path) {
            Ok(_) => Box::pin(std::future::ready(Ok(()))),
            Err(e) => Box::pin(std::future::ready(Err(map_vfs_error(e)))),
        }
    }

    fn rename<'a>(&'a self, from: &'a DavPath, to: &'a DavPath) -> FsFuture<'a, ()> {
        let from_path = match self.resolve_path(from) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        let to_path = match self.resolve_path(to) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        if let Err(e) = self.check_acl(&from_path, crate::vfs::VfsAceMask::Delete) {
            return Box::pin(std::future::ready(Err(e)));
        }
        if let Err(e) = self.check_acl(&to_path, crate::vfs::VfsAceMask::AddFile) {
            return Box::pin(std::future::ready(Err(e)));
        }
        let from_key = self.rel_key(from);
        let to_key = self.rel_key(to);
        let props_data = self.props_data.clone();
        match self.vfs.rename(&from_path, &to_path) {
            Ok(_) => {
                if let Ok(mut map) = props_data.write() {
                    if let Some(props) = map.remove(&from_key) {
                        map.insert(to_key, props);
                    }
                }
                Box::pin(std::future::ready(Ok(())))
            }
            Err(e) => Box::pin(std::future::ready(Err(map_vfs_error(e)))),
        }
    }

    fn set_accessed<'a>(&'a self, path: &'a DavPath, tm: SystemTime) -> FsFuture<'a, ()> {
        let resolved = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        match self.vfs.set_atime(&resolved, tm) {
            Ok(_) => Box::pin(std::future::ready(Ok(()))),
            Err(e) => Box::pin(std::future::ready(Err(map_vfs_error(e)))),
        }
    }

    fn set_modified<'a>(&'a self, path: &'a DavPath, tm: SystemTime) -> FsFuture<'a, ()> {
        let resolved = match self.resolve_path(path) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        match self.vfs.set_mtime(&resolved, tm) {
            Ok(_) => Box::pin(std::future::ready(Ok(()))),
            Err(e) => Box::pin(std::future::ready(Err(map_vfs_error(e)))),
        }
    }

    fn get_props<'a>(&'a self, path: &'a DavPath, _do_content: bool) -> FsFuture<'a, Vec<DavProp>> {
        let key = self.rel_key(path);
        let data = self.props_data.clone();
        Box::pin(async move {
            let map = match data.read() {
                Ok(guard) => guard,
                Err(e) => {
                    log::warn!("props_data RwLock poisoned in get_props, recovering");
                    e.into_inner()
                }
            };
            let mut props = map.get(&key).cloned().unwrap_or_default();
            if !props.iter().any(Self::is_acl_prop) {
                props.push(Self::acl_prop());
            }
            Ok(props)
        })
    }

    fn get_prop<'a>(&'a self, path: &'a DavPath, prop: DavProp) -> FsFuture<'a, Vec<u8>> {
        let key = self.rel_key(path);
        let data = self.props_data.clone();
        Box::pin(async move {
            if Self::is_acl_prop(&prop) {
                return Ok(Self::empty_acl_xml());
            }
            let map = match data.read() {
                Ok(guard) => guard,
                Err(e) => {
                    log::warn!("props_data RwLock poisoned in get_prop, recovering");
                    e.into_inner()
                }
            };
            if let Some(props) = map.get(&key) {
                for p in props {
                    if p.name == prop.name
                        && p.namespace == prop.namespace
                    {
                        return Ok(p.xml.clone().unwrap_or_default());
                    }
                }
            }
            Err(FsError::NotFound)
        })
    }

    fn patch_props<'a>(
        &'a self,
        path: &'a DavPath,
        patch: Vec<(bool, DavProp)>,
    ) -> FsFuture<'a, Vec<(StatusCode, DavProp)>> {
        let key = self.rel_key(path);
        let data = self.props_data.clone();
        let vfs = self.vfs.clone_boxed();
        let root = self.root.clone();
        Box::pin(async move {
            let results: Vec<(StatusCode, DavProp)> = {
                let mut map = match data.write() {
                    Ok(guard) => guard,
                    Err(e) => {
                        log::warn!("props_data RwLock poisoned in patch_props, recovering");
                        e.into_inner()
                    }
                };
                patch
                    .into_iter()
                    .map(|(set, prop)| {
                        let code = if set {
                            map.entry(key.clone()).or_default().push(prop.clone());
                            StatusCode::OK
                        } else {
                            if let Some(props) = map.get_mut(&key) {
                                props.retain(|p| {
                                    p.name != prop.name || p.namespace != prop.namespace
                                });
                            }
                            StatusCode::NO_CONTENT
                        };
                        (code, prop)
                    })
                    .collect()
            };
            let entries: HashMap<String, Vec<DeadPropEntry>> = {
                let map = match data.read() {
                    Ok(guard) => guard,
                    Err(e) => {
                        log::warn!("props_data RwLock poisoned in patch_props persistence, recovering");
                        e.into_inner()
                    }
                };
                map.iter()
                    .filter(|(_, v)| !v.is_empty())
                    .map(|(k, v)| (k.clone(), v.iter().map(DeadPropEntry::from).collect()))
                    .collect()
            };
            if let Ok(json) = serde_json::to_string(&entries) {
                let path = root.join(".webdav_props.json");
                let _ = tokio::task::spawn_blocking(move || {
                    let flags = OpenFlags::new().write().create().truncate().mode(0o644);
                    match vfs.open_file(&path, &flags) {
                        Ok(mut file) => {
                            if let Err(e) = file.write_all(json.as_bytes()) {
                                log::warn!("patch_props write_all failed: {:?}", e);
                            }
                            if let Err(e) = file.flush() {
                                log::warn!("patch_props flush failed: {:?}", e);
                            }
                        }
                        Err(e) => {
                            log::warn!("patch_props open_file failed: {:?}", e);
                        }
                    }
                }).await;
            }
            Ok(results)
        })
    }

    fn have_props<'a>(&'a self, _path: &'a DavPath) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
        Box::pin(std::future::ready(true))
    }

    fn get_quota(&'_ self) -> FsFuture<'_, (u64, Option<u64>)> {
        let root = self.root.clone();
        let vfs = self.vfs.clone_boxed();
        Box::pin(async move {
            let used = vfs.get_quota_usage(&root);
            let total = vfs.get_quota(&root);
            match (used, total) {
                (Ok(usage), Ok(quota)) => {
                    let limit = if quota.space_limit > 0 { Some(quota.space_limit) } else { None };
                    Ok((usage.space_used, limit))
                }
                (Err(VfsError::NotFound(_)), _) | (_, Err(VfsError::NotFound(_))) => {
                    Err(FsError::NotFound)
                }
                (Err(VfsError::Unsupported(_)), _) | (_, Err(VfsError::Unsupported(_))) => {
                    Err(FsError::NotImplemented)
                }
                _ => Err(FsError::GeneralFailure),
            }
        })
    }

    fn copy<'a>(&'a self, from: &'a DavPath, to: &'a DavPath) -> FsFuture<'a, ()> {
        let from_path = match self.resolve_path(from) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };
        let to_path = match self.resolve_path(to) {
            Ok(p) => p,
            Err(e) => return Box::pin(std::future::ready(Err(e))),
        };

        let from_stat = match self.vfs.stat(&from_path) {
            Ok(s) => s,
            Err(_) => return Box::pin(std::future::ready(Err(FsError::NotFound))),
        };

        let from_key = self.rel_key(from);
        let to_key = self.rel_key(to);
        let props_data = self.props_data.clone();

        if from_stat.is_dir {
            let vfs_arc = self.vfs.clone_boxed();
            Box::pin(async move {
                if let Err(e) = copy_dir_recursive_impl(&*vfs_arc, &from_path, &to_path) {
                    return Err(match e {
                        VfsError::NotFound(_) => FsError::NotFound,
                        VfsError::PermissionDenied(_) => FsError::Forbidden,
                        _ => FsError::NotImplemented,
                    });
                }
                if let Ok(map) = props_data.read() {
                    if let Some(props) = map.get(&from_key) {
                        if let Ok(mut map) = props_data.write() {
                            map.insert(to_key, props.clone());
                        }
                    }
                }
                Ok(())
            })
        } else {
            match self.vfs.copy(&from_path, &to_path) {
                Ok(_) => {
                    if let Ok(map) = props_data.read() {
                        if let Some(props) = map.get(&from_key) {
                            if let Ok(mut map) = props_data.write() {
                                map.insert(to_key, props.clone());
                            }
                        }
                    }
                    Box::pin(std::future::ready(Ok(())))
                }
                Err(VfsError::NotFound(_)) => Box::pin(std::future::ready(Err(FsError::NotFound))),
                Err(_) => Box::pin(std::future::ready(Err(FsError::NotImplemented))),
            }
        }
    }
}

/// Recursive directory copy via VfsBackend
fn copy_dir_recursive_impl(
    vfs: &dyn VfsBackend,
    src: &Path,
    dst: &Path,
) -> Result<(), VfsError> {
    vfs.create_dir_all(dst, 0o755)?;
    let entries = vfs.read_dir(src)?;
    for entry in entries {
        let src_entry = src.join(&entry.name);
        let dst_entry = dst.join(&entry.name);
        if entry.stat.is_dir {
            copy_dir_recursive_impl(vfs, &src_entry, &dst_entry)?;
        } else {
            vfs.copy(&src_entry, &dst_entry)?;
        }
    }
    Ok(())
}

pub fn create_webdav_handler(
    vfs: Box<dyn VfsBackend>,
    root: PathBuf,
    upload_hook: Option<Arc<UploadHook>>,
    user_uuid: String,
) -> dav_server::DavHandler {
    create_webdav_handler_inner(vfs, root, upload_hook, user_uuid, None, None)
}

pub fn create_webdav_handler_with_versioning(
    vfs: Box<dyn VfsBackend>,
    root: PathBuf,
    upload_hook: Option<Arc<UploadHook>>,
    user_uuid: String,
    versioning: Arc<WebDavVersioning>,
) -> dav_server::DavHandler {
    create_webdav_handler_inner(vfs, root, upload_hook, user_uuid, Some(versioning), None)
}

pub fn create_webdav_handler_persisted(
    vfs: Box<dyn VfsBackend>,
    root: PathBuf,
    upload_hook: Option<Arc<UploadHook>>,
    user_uuid: String,
    versioning: Option<Arc<WebDavVersioning>>,
    locks_file: PathBuf,
) -> dav_server::DavHandler {
    create_webdav_handler_inner(vfs, root, upload_hook, user_uuid, versioning, Some(locks_file))
}

fn create_webdav_handler_inner(
    vfs: Box<dyn VfsBackend>,
    root: PathBuf,
    upload_hook: Option<Arc<UploadHook>>,
    user_uuid: String,
    versioning: Option<Arc<WebDavVersioning>>,
    locks_file: Option<PathBuf>,
) -> dav_server::DavHandler {
    let dav_fs = match versioning {
        Some(v) => VfsDavFs::with_versioning(vfs, root, upload_hook, user_uuid, v),
        None => VfsDavFs::new(vfs, root, upload_hook, user_uuid),
    };
    let locksystem: Box<dyn DavLockSystem> = match locks_file {
        Some(path) => PersistedLs::new(path),
        None => MemLs::new(),
    };
    dav_server::DavHandler::builder()
        .filesystem(dav_fs)
        .locksystem(locksystem)
        .strip_prefix("")
        .build_handler()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::vfs::local_fs::LocalFs;
    use dav_server::davpath::DavPath;
    use std::fs;
    use tempfile::TempDir;

    #[test]
    fn test_dead_prop_entry_roundtrip() {
        let prop = DavProp {
            name: "displayname".to_string(),
            prefix: Some("D".to_string()),
            namespace: Some("DAV:".to_string()),
            xml: Some(b"<D:displayname>test.txt</D:displayname>".to_vec()),
        };
        let entry = DeadPropEntry::from(&prop);
        let restored = DavProp::from(entry);
        assert_eq!(restored.name, "displayname");
        assert_eq!(restored.prefix, Some("D".to_string()));
        assert_eq!(restored.namespace, Some("DAV:".to_string()));
        assert!(restored.xml.is_some());
    }

    #[test]
    fn test_dead_prop_entry_serde() {
        let prop = DavProp {
            name: "getcontenttype".to_string(),
            prefix: Some("D".to_string()),
            namespace: Some("DAV:".to_string()),
            xml: Some(b"text/plain".to_vec()),
        };
        let entry = DeadPropEntry::from(&prop);
        let json = serde_json::to_string(&entry).unwrap();
        let deserialized: DeadPropEntry = serde_json::from_str(&json).unwrap();
        let restored = DavProp::from(deserialized);
        assert_eq!(restored.name, "getcontenttype");
        assert_eq!(restored.xml.unwrap(), b"text/plain".to_vec());
    }

    #[test]
    fn test_dead_prop_entry_no_xml() {
        let prop = DavProp {
            name: "resourcetype".to_string(),
            prefix: None,
            namespace: None,
            xml: None,
        };
        let entry = DeadPropEntry::from(&prop);
        let json = serde_json::to_string(&entry).unwrap();
        let deserialized: DeadPropEntry = serde_json::from_str(&json).unwrap();
        assert!(deserialized.xml.is_none());
    }

    #[test]
    fn test_vfs_dav_meta_data_from_stat() {
        let stat = crate::vfs::VfsStat {
            size: 1024,
            mode: 0o644,
            uid: 501,
            gid: 20,
            atime: std::time::UNIX_EPOCH,
            mtime: std::time::UNIX_EPOCH,
            is_dir: false,
            is_symlink: false,
        };
        let meta = VfsDavMetaData::from_stat(&stat);
        assert_eq!(meta.len(), 1024);
        assert!(!meta.is_dir());
    }

    #[test]
    fn test_vfs_dav_meta_data_dir() {
        let stat = crate::vfs::VfsStat {
            size: 0,
            mode: 0o755,
            uid: 501,
            gid: 20,
            atime: std::time::UNIX_EPOCH,
            mtime: std::time::UNIX_EPOCH,
            is_dir: true,
            is_symlink: false,
        };
        let meta = VfsDavMetaData::from_stat(&stat);
        assert_eq!(meta.len(), 0);
        assert!(meta.is_dir());
    }

    #[test]
    fn test_resolve_path_traversal_rejected() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().to_path_buf();
        let vfs = LocalFs::new();
        let dav_fs = VfsDavFs::new(Box::new(vfs), root.clone(), None, "test".to_string());

        // DavPath rejects paths containing ".." outright (ForbiddenPath),
        // so the earliest our resolve_path can catch is absolute-path-like
        // paths that escape root via canonicalize mismatch.
        // Create a file outside root and try to access it.
        let outside = TempDir::new().unwrap();
        let secret = outside.path().join("secret.txt");
        fs::write(&secret, "data").unwrap();

        // An absolute path that doesn't start with root should be rejected
        // by the canonicalize check inside resolve_path.
        // We skip DavPath and call the internal method directly:
        let bad_relative = std::path::Path::new("/etc/passwd");
        // resolve_path accepts DavPath, so test the internal logic:
        let full = root.join(bad_relative);
        let root_canonical = root.canonicalize().unwrap();
        if let Ok(canonical) = full.canonicalize() {
            assert!(!canonical.starts_with(&root_canonical));
        } else {
            // Path doesn't exist, which means it'll be caught by the
            // parent validation (or fail to canonicalize), which is also correct
        }
    }

    #[test]
    fn test_resolve_path_normal() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().to_path_buf();
        fs::write(root.join("test.txt"), "hello").unwrap();

        let vfs = LocalFs::new();
        let dav_fs = VfsDavFs::new(Box::new(vfs), root.clone(), None, "test".to_string());

        // DavPath is always absolute (starts with /) — it's a URL path convention.
        // resolve_path uses as_rel_ospath() to strip the leading / before joining with root.
        let dp = DavPath::new("/test.txt").unwrap();
        let resolved = dav_fs.resolve_path(&dp).unwrap();
        assert!(resolved.exists());
        assert_eq!(resolved.file_name().unwrap().to_str(), Some("test.txt"));
        assert!(resolved.starts_with(root.canonicalize().unwrap()));
    }

    #[test]
    fn test_resolve_path_nonexistent_parent_ok() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().to_path_buf();
        let vfs = LocalFs::new();
        let dav_fs = VfsDavFs::new(Box::new(vfs), root.clone(), None, "test".to_string());

        // Non-existent file in existing root — should succeed (parent=root is valid)
        let dp = DavPath::new("/new_file.txt").unwrap();
        let result = dav_fs.resolve_path(&dp);
        assert!(result.is_ok());

        // Non-existent file in non-existent subdir — should fail (parent doesn't exist)
        let dp2 = DavPath::new("/nonexistent_dir/file.txt").unwrap();
        let result2 = dav_fs.resolve_path(&dp2);
        assert!(result2.is_err());
    }

    #[test]
    fn test_copy_dir_recursive() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("src");
        let dst = tmp.path().join("dst");

        fs::create_dir_all(root.join("subdir")).unwrap();
        fs::write(root.join("a.txt"), "aaa").unwrap();
        fs::write(root.join("subdir").join("b.txt"), "bbb").unwrap();

        let vfs = LocalFs::new();
        copy_dir_recursive_impl(&vfs, &root, &dst).unwrap();

        assert!(dst.join("a.txt").exists());
        assert!(dst.join("subdir").join("b.txt").exists());
        assert_eq!(fs::read_to_string(dst.join("a.txt")).unwrap(), "aaa");
    }

    #[test]
    fn test_remove_dir_all() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().join("toremove");
        fs::create_dir_all(dir.join("sub").join("deep")).unwrap();
        fs::write(dir.join("f1.txt"), "data").unwrap();
        fs::write(dir.join("sub").join("f2.txt"), "data").unwrap();

        let vfs = LocalFs::new();
        vfs.remove_dir_all(&dir).unwrap();

        assert!(!dir.exists());
    }

    #[test]
    fn test_create_dir_all_nested() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("base");
        let nested = root.join("a").join("b").join("c");

        let vfs = LocalFs::new();
        vfs.create_dir_all(&nested, 0o755).unwrap();

        assert!(nested.exists());
        assert!(nested.is_dir());
    }

    #[test]
    fn test_set_times() {
        let tmp = TempDir::new().unwrap();
        let file = tmp.path().join("time_test.txt");
        fs::write(&file, "data").unwrap();

        let vfs = LocalFs::new();
        let atime = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1000000);
        let mtime = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(2000000);

        vfs.set_times(&file, atime, mtime).unwrap();

        let stat = vfs.stat(&file).unwrap();
        assert_eq!(stat.atime, atime);
        assert_eq!(stat.mtime, mtime);
    }

    #[test]
    fn test_dead_props_store_and_load() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().to_path_buf();
        let vfs = LocalFs::new();
        let dav_fs = VfsDavFs::new(Box::new(vfs), root.clone(), None, "test".to_string());

        let prop = DavProp {
            name: "custom".to_string(),
            prefix: None,
            namespace: None,
            xml: Some(b"value".to_vec()),
        };

        // Simulate patch_props by directly inserting
        {
            let mut map = dav_fs.props_data.write().unwrap();
            map.entry("/test".to_string()).or_default().push(prop.clone());
        }
        dav_fs.save_props();

        // Load from file into a new VfsDavFs
        let vfs2 = LocalFs::new();
        let dav_fs2 = VfsDavFs::new(Box::new(vfs2), root.clone(), None, "test".to_string());
        let map = dav_fs2.props_data.read().unwrap();
        let props = map.get("/test");
        assert!(props.is_some());
        assert_eq!(props.unwrap().len(), 1);
        assert_eq!(props.unwrap()[0].name, "custom");
    }
}

#[cfg(test)]
mod integration_tests {
    use axum::body::Body as ReqBody;
    use dav_server::body::Body as ResBody;
    use dav_server::DavHandler;
    use dav_server::memls::MemLs;
    use futures_util::TryStreamExt;
    use http::{Method, StatusCode};
    use tempfile::TempDir;

    use super::VfsDavFs;
    use crate::vfs::local_fs::LocalFs;

    fn make_handler(tmp: &TempDir) -> DavHandler {
        let root = tmp.path().to_path_buf();
        let vfs = Box::new(LocalFs::new());
        let dav_fs = VfsDavFs::new(vfs, root, None, "test".to_string());
        DavHandler::builder()
            .filesystem(dav_fs)
            .locksystem(MemLs::new())
            .build_handler()
    }

    async fn collect_body(res: http::Response<ResBody>) -> (StatusCode, Vec<u8>) {
        let (parts, body) = res.into_parts();
        let bytes = body
            .try_fold(Vec::new(), |mut acc, chunk| async move {
                acc.extend_from_slice(&chunk);
                Ok(acc)
            })
            .await
            .unwrap();
        (parts.status, bytes)
    }

    #[tokio::test]
    async fn test_put_get_roundtrip() {
        let tmp = TempDir::new().unwrap();
        let handler = make_handler(&tmp);

        // PUT
        let req = http::Request::builder()
            .method(Method::PUT)
            .uri("/hello.txt")
            .body(ReqBody::from("Hello, World!"))
            .unwrap();
        let res = handler.handle(req).await;
        let status = res.status();
        assert!(
            status == StatusCode::CREATED || status == StatusCode::NO_CONTENT,
            "Expected CREATED (201) or NO_CONTENT (204), got {}",
            status
        );

        // GET
        let req = http::Request::builder()
            .method(Method::GET)
            .uri("/hello.txt")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        let (status, body) = collect_body(res).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, b"Hello, World!");
    }

    #[tokio::test]
    async fn test_propfind() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("a.txt"), "aaa").unwrap();
        std::fs::write(tmp.path().join("b.txt"), "bbb").unwrap();
        std::fs::create_dir(tmp.path().join("subdir")).unwrap();
        let handler = make_handler(&tmp);

        let req = http::Request::builder()
            .method(Method::from_bytes(b"PROPFIND").unwrap())
            .uri("/")
            .header("Depth", "1")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert_eq!(res.status(), StatusCode::MULTI_STATUS);
        let (_, body) = collect_body(res).await;
        let xml = String::from_utf8(body).unwrap();
        assert!(xml.contains("a.txt"), "PROPFIND response should contain a.txt");
        assert!(xml.contains("b.txt"), "PROPFIND response should contain b.txt");
        assert!(xml.contains("subdir"), "PROPFIND response should contain subdir");
    }

    #[tokio::test]
    async fn test_mkcol_delete() {
        let tmp = TempDir::new().unwrap();
        let handler = make_handler(&tmp);

        // MKCOL
        let req = http::Request::builder()
            .method(Method::from_bytes(b"MKCOL").unwrap())
            .uri("/mydir")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert!(res.status().is_success(), "MKCOL failed: {}", res.status());
        assert!(tmp.path().join("mydir").is_dir(), "Directory should exist");

        // DELETE
        let req = http::Request::builder()
            .method(Method::DELETE)
            .uri("/mydir")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert!(res.status().is_success(), "DELETE failed: {}", res.status());
        assert!(!tmp.path().join("mydir").exists(), "Directory should be removed");
    }

    #[tokio::test]
    async fn test_copy_move() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("src.txt"), "data").unwrap();
        let handler = make_handler(&tmp);

        // COPY
        let req = http::Request::builder()
            .method(Method::from_bytes(b"COPY").unwrap())
            .uri("/src.txt")
            .header("Destination", "/dst.txt")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert!(res.status().is_success(), "COPY failed: {}", res.status());
        assert_eq!(
            std::fs::read_to_string(tmp.path().join("dst.txt")).unwrap(),
            "data",
            "Copied file content should match"
        );

        // MOVE
        let req = http::Request::builder()
            .method(Method::from_bytes(b"MOVE").unwrap())
            .uri("/src.txt")
            .header("Destination", "/moved.txt")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert!(res.status().is_success(), "MOVE failed: {}", res.status());
        assert!(!tmp.path().join("src.txt").exists(), "Source should not exist after MOVE");
        assert_eq!(
            std::fs::read_to_string(tmp.path().join("moved.txt")).unwrap(),
            "data",
            "Moved file content should match"
        );
    }

    #[tokio::test]
    async fn test_lock_unlock() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("locked.txt"), "content").unwrap();
        let handler = make_handler(&tmp);

        // LOCK
        let lock_body = r#"<?xml version="1.0" encoding="utf-8"?>
<D:lockinfo xmlns:D="DAV:">
  <D:lockscope><D:exclusive/></D:lockscope>
  <D:locktype><D:write/></D:locktype>
  <D:owner><D:href>testuser</D:href></D:owner>
</D:lockinfo>"#;
        let req = http::Request::builder()
            .method(Method::from_bytes(b"LOCK").unwrap())
            .uri("/locked.txt")
            .header("Content-Type", "text/xml; charset=utf-8")
            .body(ReqBody::from(lock_body))
            .unwrap();
        let res = handler.handle(req).await;
        assert_eq!(res.status(), StatusCode::OK, "LOCK should return 200 OK");

        let lock_token = res
            .headers()
            .get("Lock-Token")
            .and_then(|v| v.to_str().ok())
            .map(String::from);
        assert!(
            lock_token.is_some(),
            "LOCK response should have Lock-Token header"
        );

        // UNLOCK
        let req = http::Request::builder()
            .method(Method::from_bytes(b"UNLOCK").unwrap())
            .uri("/locked.txt")
            .header("Lock-Token", lock_token.as_ref().unwrap())
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert_eq!(
            res.status(),
            StatusCode::NO_CONTENT,
            "UNLOCK should return 204 NO_CONTENT"
        );
    }

    #[tokio::test]
    async fn test_etag_header() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("etagtest.txt"), "etag content").unwrap();
        let handler = make_handler(&tmp);

        let req = http::Request::builder()
            .method(Method::GET)
            .uri("/etagtest.txt")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert_eq!(res.status(), StatusCode::OK);
        assert!(
            res.headers().contains_key("ETag"),
            "GET response should include ETag header"
        );
    }

    #[tokio::test]
    async fn test_acl_property() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("acl_test.txt"), "acl content").unwrap();
        let handler = make_handler(&tmp);

        // PROPFIND with DAV:acl property requested
        let req_body = r#"<?xml version="1.0" encoding="utf-8"?>
<D:propfind xmlns:D="DAV:">
  <D:prop>
    <D:acl/>
    <D:resourcetype/>
    <D:displayname/>
  </D:prop>
</D:propfind>"#;
        let req = http::Request::builder()
            .method(Method::from_bytes(b"PROPFIND").unwrap())
            .uri("/acl_test.txt")
            .header("Depth", "0")
            .header("Content-Type", "text/xml; charset=utf-8")
            .body(ReqBody::from(req_body))
            .unwrap();
        let res = handler.handle(req).await;
        assert_eq!(res.status(), StatusCode::MULTI_STATUS);
        let (_, body) = collect_body(res).await;
        let xml = String::from_utf8(body).unwrap();
        assert!(xml.contains("D:acl"), "PROPFIND response should contain D:acl element");
        assert!(xml.contains("D:all"), "ACL should grant all privileges to all principals");
    }

    #[tokio::test]
    async fn test_range_request() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("range_test.txt"), "0123456789abcdefghij").unwrap();
        let handler = make_handler(&tmp);

        // Range request: bytes=5-10
        let req = http::Request::builder()
            .method(Method::GET)
            .uri("/range_test.txt")
            .header("Range", "bytes=5-10")
            .body(ReqBody::from(""))
            .unwrap();
        let res = handler.handle(req).await;
        assert_eq!(res.status(), StatusCode::PARTIAL_CONTENT, "Range request should return 206");
        assert!(
            res.headers().contains_key("Content-Range"),
            "Range response should include Content-Range header"
        );
        let (_, body) = collect_body(res).await;
        let content = String::from_utf8(body).unwrap();
        assert_eq!(content, "56789a", "Range bytes=5-10 should return '56789a' (6 bytes)");
    }
}