//! WebDAV endpoint.
//!
//! Two mounts, both served by the same [`FbFs`]:
//!
//! * `{DAV}` — a signed-in user's roots. Each root is a child collection of a
//!   synthetic top-level directory, so one mount covers every root the user
//!   has. Basic takes either the account password or one of the account's app
//!   passwords.
//! * `{DAV_SHARE}/{token}` — one public share, mounted at its own root.
//!
//! All filesystem access goes through [`crate::fs`], so a mount inherits the
//! same containment and the same symlink handling the JSON API has.
//!
//! The protocol itself (PROPFIND, the 207 multistatus, `Depth`, `Destination`,
//! `Overwrite`, conditional headers) is `dav-server`'s job. This module only
//! authenticates the request, decides which roots it may see, and maps dav
//! paths onto real ones.

use std::collections::HashMap;
use std::io::SeekFrom;
use std::path::{Path, PathBuf};
use std::sync::{Arc, LazyLock, Mutex, Weak};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use api_types::{AuthMode, DAV, DAV_SHARE, Mode};
use axum::body::Body;
use axum::extract::State;
use axum::http::header::{HeaderMap, WWW_AUTHENTICATE};
use axum::http::{Request, Response, StatusCode};
use axum::response::IntoResponse;
use bytes::{Buf, Bytes};
use dav_server::DavConfig;
use dav_server::davpath::{DavPath, ParseError};
use dav_server::fs::{
    DavDirEntry, DavFile, DavMetaData, FsError, FsFuture, FsResult, FsStream, GuardedFileSystem,
    OpenOptions, ReadDirMeta,
};
use dav_server::ls::{DavLock, DavLockSystem, LsFuture};
use dav_server::memls::MemLs;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};

use crate::api::common::{display_name, session_auth};
use crate::auth;
use crate::db::RootRow;
use crate::error::AppState;

/// The `WWW-Authenticate` realm. Clients show it in their password prompt.
const REALM: &str = "dovenest";

// ---------------------------------------------------------------------------
// Routes
// ---------------------------------------------------------------------------

/// `{DAV}` and everything under it: the signed-in user's roots.
///
/// A browser session cookie is accepted, but the usual caller is a mount
/// client, which only speaks HTTP Basic.
pub async fn user(State(state): State<Arc<AppState>>, req: Request<Body>) -> Response<Body> {
    let Some((user_id, principal)) = authenticate(&state, req.headers()).await else {
        return challenge();
    };
    let Ok(roots) = state.db.user_roots(user_id).await else {
        return StatusCode::INTERNAL_SERVER_ERROR.into_response();
    };
    // The admin pseudo-root (the whole server root, read-only) is deliberately
    // not mounted: `session_auth` does not add it, and a mount that silently
    // contained a second copy of every other root would be confusing.
    let mount = Mount {
        roots: Arc::new(root_segments(&state, roots)),
        flat: false,
    };
    serve(state, req, DAV.to_string(), principal, mount).await
}

/// `{DAV_SHARE}/{token}` and everything under it: one public share.
///
/// Folder shares only. A file share has no collection to mount, and its one
/// file is already a plain `GET` away on the share page.
pub async fn share(State(state): State<Arc<AppState>>, req: Request<Body>) -> Response<Body> {
    let Some(token) = share_token(req.uri().path()) else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let row = match state.db.share_by_token(&token).await {
        Ok(Some(row)) => row,
        Ok(None) => return StatusCode::NOT_FOUND.into_response(),
        Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(),
    };
    if row.is_expired() {
        return StatusCode::GONE.into_response();
    }
    if row.is_file {
        return StatusCode::NOT_FOUND.into_response();
    }
    // A protected share takes its password over Basic, with the user name
    // ignored. There is no account behind a share link to name.
    if let Some(hash) = row.password_hash.clone() {
        let Some((_, password)) = auth::basic_credentials(req.headers()) else {
            return challenge();
        };
        let (pw, id, tok) = (password.clone(), row.id, token.clone());
        let ok = auth::verify_cached(row.id, "", &password, move || async move {
            // Throttled like `POST /api/share/{token}/unlock`, keyed the same
            // way, so a mount client is not the cheap way to guess.
            auth::throttle(&tok).await;
            let ok = auth::verify_password_async(&pw, &hash).await;
            auth::record_login(&tok, ok);
            ok.then_some(id)
        })
        .await;
        if ok.is_none() {
            return challenge();
        }
    }
    let root = RootRow {
        id: row.id,
        path: row.target.clone(),
        mode: row.mode,
    };
    let mount = Mount {
        roots: Arc::new(vec![(String::new(), root)]),
        flat: true,
    };
    let prefix = format!("{DAV_SHARE}/{token}");
    serve(state, req, prefix, format!("share-{}", row.id), mount).await
}

/// Whether the request path still addresses this mount after `dav-server` has
/// normalized it: percent-decoded, `.` and `..` resolved, slashes merged.
///
/// Runs the same two steps as the handler, so it rejects nothing the handler
/// would accept. The other parse errors (`InvalidPath`, `ForbiddenPath`) are
/// left to the handler, which answers those with a 4xx of its own.
fn path_in_mount(path: &str, prefix: &str) -> bool {
    // `OPTIONS *` has no leading slash. The handler answers it.
    if !path.starts_with('/') {
        return true;
    }
    !matches!(
        DavPath::new(path).and_then(|mut p| p.set_prefix(prefix)),
        Err(ParseError::PrefixMismatch)
    )
}

/// The `Destination` header as a URL path, the way `dav-server`'s private
/// header parser reads it: a path as-is, a full URL (what mount clients send)
/// reduced to its path. `None` for a missing or unparseable header.
fn destination_path(headers: &HeaderMap) -> Option<String> {
    let raw = headers.get("destination")?.to_str().ok()?;
    if raw.starts_with('/') {
        return Some(raw.to_string());
    }
    raw.parse::<axum::http::Uri>()
        .ok()
        .map(|u| u.path().to_string())
}

/// The token out of the *raw* URL path.
///
/// Not axum's decoded wildcard: `DavPath` keeps the raw path, and
/// `strip_prefix` byte-compares against it. A decoded `<token>%2Fx` would
/// yield a prefix that the dav path does not start with.
fn share_token(path: &str) -> Option<String> {
    let rest = path.strip_prefix(DAV_SHARE)?.strip_prefix('/')?;
    let token = rest.split('/').next().unwrap_or_default();
    (!token.is_empty()).then(|| token.to_string())
}

/// Hand the request to `dav-server` and, afterwards, keep the share table in
/// step with the filesystem.
///
/// The handler is built here rather than once at startup because `prefix`
/// differs per share mount, and `dav-server` only allows a per-request config
/// override on the unguarded handler. Building it is cheap: an `Arc::new` and
/// two `Arc`-backed trait-object clones.
async fn serve(
    state: Arc<AppState>,
    req: Request<Body>,
    prefix: String,
    principal: String,
    mount: Mount,
) -> Response<Body> {
    // `dav-server` refuses an escaping path too, but with `502 Bad Gateway`
    // (`DavError::IllegalPath`), which reads as a broken upstream. A `..` that
    // stays inside the mount is already a `403`, so answer this the same way.
    // The same for a COPY or MOVE `Destination`, which the handler normalizes
    // the same way. A missing or malformed header stays with the handler.
    let dest_in_mount =
        destination_path(req.headers()).is_none_or(|dest| path_in_mount(&dest, &prefix));
    if !path_in_mount(req.uri().path(), &prefix) || !dest_in_mount {
        return StatusCode::FORBIDDEN.into_response();
    }

    // `dav-server` reads every other body whole, up to this size, and its
    // XML parser recurses once per level: a deep body overflows the stack.
    let req = match req.method().as_str() {
        "PUT" | "PATCH" => req,
        _ => {
            let (parts, body) = req.into_parts();
            let Ok(body) = axum::body::to_bytes(body, 65_536).await else {
                return StatusCode::PAYLOAD_TOO_LARGE.into_response();
            };
            if pimdav::xml::too_deep(&body) {
                return StatusCode::BAD_REQUEST.into_response();
            }
            Request::from_parts(parts, Body::from(body))
        }
    };

    // Resolved *before* the operation, while the item still exists: once
    // DELETE or MOVE has run there is no path left to look a share up by.
    let vacating = matches!(req.method().as_str(), "DELETE" | "MOVE");
    let vacated = if vacating {
        dav_target(&state, &mount, &prefix, &req)
    } else {
        None
    };

    let handler = DavConfig::<Mount>::new()
        .filesystem(Box::new(FbFs {
            state: state.clone(),
        }))
        // The handler is rebuilt per request, the lock tree must not be.
        .locksystem(Box::new(locks_for(&principal)))
        // Its only effect in `dav-server` is picking the `ReadDirMeta` for
        // PROPFIND. `false` keeps listings on the followed metadata.
        .hide_symlinks(false)
        // Off by default in `dav-server`: without it a plain `GET` of any
        // collection answers 405, so the mount is unreadable in a browser.
        .autoindex(true)
        .strip_prefix(prefix)
        .build_handler();

    let mut resp = handler.handle_guarded(req, principal, mount).await;
    crate::api::sandbox_scriptable(&mut resp);

    // One revoke for the whole request. Doing it inside the filesystem would
    // fire a query per removed item, and a recursive DELETE walks the tree.
    if let Some(abs) = vacated
        && resp.status().is_success()
    {
        crate::api::files::revoke_shares_at(&state, &abs).await;
    }
    resp.map(Body::new)
}

/// The absolute path a request addresses, if it resolves to one today.
fn dav_target(
    state: &AppState,
    mount: &Mount,
    prefix: &str,
    req: &Request<Body>,
) -> Option<PathBuf> {
    // `DavPath::new`, not `from_uri`: the latter keeps the raw bytes, so an
    // encoded path (`a%20b.txt`) would never resolve and the share would
    // outlive the file it named.
    let mut path = DavPath::new(req.uri().path()).ok()?;
    path.set_prefix(prefix).ok()?;
    let (root, rel) = item(&path, mount).ok()?;
    // `resolve_entry`, matching the operations this is predicting. Following
    // the last component would name a symlink's target, so deleting a link
    // would revoke a share on a file that is still there.
    crate::fs::resolve_entry(&state.root, &root.path, &rel).ok()
}

/// 401 with the Basic challenge every mount client needs to see before it
/// will send credentials at all.
pub(crate) fn challenge() -> Response<Body> {
    (
        StatusCode::UNAUTHORIZED,
        // RFC 7617: without a charset, clients may send a non-ASCII password
        // as Latin-1.
        [(
            WWW_AUTHENTICATE,
            format!("Basic realm=\"{REALM}\", charset=\"UTF-8\""),
        )],
    )
        .into_response()
}

// ---------------------------------------------------------------------------
// Authentication
// ---------------------------------------------------------------------------

/// Resolve the caller to a user id and name.
pub(crate) async fn authenticate(state: &AppState, headers: &HeaderMap) -> Option<(i64, String)> {
    // A browser hitting the mount already has a session; take it and skip
    // Argon2 entirely.
    if auth::parse_session_cookie(headers).is_some()
        && let Ok((user, _)) = session_auth(headers, state).await
    {
        return Some((user.id, user.name));
    }

    let (name, password) = auth::basic_credentials(headers)?;

    // The Basic user name is ignored: the secret already names the account.
    match state
        .db
        .user_by_app_password(&auth::app_password_hash(&password))
        .await
    {
        Ok(Some(user)) => return Some((user.id, user.name)),
        Ok(None) => {}
        // A lookup error 401s a valid app password and counts as a failed
        // login for that name below. Nothing else records that.
        Err(e) => tracing::warn!(error = %e, "app password lookup failed"),
    }

    // iOS 18.4 and later send `@` in the user name as `%40`. Decoded only
    // when no account has the name as sent, and before the one verify, so
    // the throttle counts one attempt.
    let name = match percent_encoding::percent_decode_str(&name).decode_utf8() {
        Ok(decoded)
            if decoded != name && matches!(state.db.find_user_by_name(&name).await, Ok(None)) =>
        {
            decoded.into_owned()
        }
        _ => name,
    };

    let id = auth::verify_cached(0, &name, &password, || {
        let (state, name, password) = (state, name.clone(), password.clone());
        async move {
            // The login route's throttle, keyed the same way, so guessing over
            // WebDAV is no cheaper than guessing over the login form.
            auth::throttle(&name).await;
            let verified = state.db.verify_password(&name, &password).await.ok()?;
            // Record what the password did, not what the rule below decides.
            // A mount on an account that requires a passkey keeps retrying,
            // and counting each retry as a failed guess would pin that name's
            // delay and lock the person out of the web login too.
            auth::record_login(&name, verified.is_some());
            // Basic carries a password and nothing else, so accepting the
            // account password here would downgrade an account that asks for
            // a passkey too. Such an account mounts with an app password,
            // which the lookup above already handled.
            verified
                .filter(|u| u.auth_mode == AuthMode::Either)
                .map(|u| u.id)
        }
    })
    .await?;
    Some((id, name))
}

// ---------------------------------------------------------------------------
// Locking
// ---------------------------------------------------------------------------

/// One lock tree per principal.
///
/// A lock is keyed by DAV URL, and a URL segment is a root's display name, so a
/// single shared tree lets two users whose roots are both named `Documents`
/// reach each other's locks. One could block the other, and `PROPFIND` hands
/// back the holder's lock token, which is enough to release that lock or write
/// through it.
///
/// The cost is that two principals sharing one physical folder do not
/// coordinate through WebDAV locks. Byte-level safety does not rest on this:
/// [`WRITE_LOCKS`] keys on the resolved path and covers every writer.
///
/// `MemLs` keeps its state in an `Arc`, so a clone shares that principal's
/// tree. Locks live in memory only, and a restart drops them all, which is
/// what a client already sees when a lock times out.
static LOCKS: LazyLock<Mutex<HashMap<String, ExpiringLs>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

fn locks_for(principal: &str) -> ExpiringLs {
    let mut g = LOCKS.lock().unwrap_or_else(|e| e.into_inner());
    g.entry(principal.to_string())
        .or_insert_with(|| ExpiringLs(*MemLs::new()))
        .clone()
}

/// Longest lock handed out, and so the longest an abandoned one blocks a file.
/// A client that still wants the file refreshes; one that crashed does not.
///
/// Matches `dav-server`'s own ceiling for an exclusive lock. It caps only the
/// two cases that arrive here uncapped, both as `None` meaning "never
/// expires": a LOCK with no `Timeout` header, and a refresh asking for
/// `Infinite`.
const LOCK_TIMEOUT: Duration = Duration::from_secs(600);

/// [`MemLs`] with lock expiry actually applied.
///
/// `MemLs` records a lock's `timeout_at` and then never looks at it again, and
/// it honours an infinite timeout request. Left alone, a client that died
/// holding an exclusive lock would block that file until the process restarts.
/// Every call here first drops the expired locks covering the path it touches,
/// and no lock is granted for longer than [`LOCK_TIMEOUT`].
#[derive(Debug, Clone)]
struct ExpiringLs(MemLs);

impl ExpiringLs {
    /// Drop the expired locks on `path` and its ancestors.
    ///
    /// Only the locks that could block an operation *on this path*. A lock on
    /// a descendant is not swept, so a deep operation can still be refused by
    /// a stale lock below it until something touches that path directly.
    async fn sweep(&self, path: &DavPath) {
        let now = SystemTime::now();
        for lock in self.0.discover(path).await {
            if lock.timeout_at.is_some_and(|t| t <= now) {
                let _ = self.0.unlock(&lock.path, &lock.token).await;
            }
        }
    }

    /// Never `None`, never longer than [`LOCK_TIMEOUT`]. `None` would mean a
    /// lock that [`sweep`](Self::sweep) can never clear.
    fn capped(timeout: Option<Duration>) -> Option<Duration> {
        Some(timeout.unwrap_or(LOCK_TIMEOUT).min(LOCK_TIMEOUT))
    }
}

impl DavLockSystem for ExpiringLs {
    fn lock(
        &self,
        path: &DavPath,
        principal: Option<&str>,
        owner: Option<&xmltree::Element>,
        timeout: Option<Duration>,
        shared: bool,
        deep: bool,
    ) -> LsFuture<'_, Result<DavLock, DavLock>> {
        // The borrows end with the call, not with the future, so clone into it.
        let (path, principal) = (path.clone(), principal.map(str::to_string));
        let owner = owner.cloned();
        Box::pin(async move {
            self.sweep(&path).await;
            self.0
                .lock(
                    &path,
                    principal.as_deref(),
                    owner.as_ref(),
                    Self::capped(timeout),
                    shared,
                    deep,
                )
                .await
        })
    }

    fn unlock(&self, path: &DavPath, token: &str) -> LsFuture<'_, Result<(), ()>> {
        let (path, token) = (path.clone(), token.to_string());
        Box::pin(async move { self.0.unlock(&path, &token).await })
    }

    fn refresh(
        &self,
        path: &DavPath,
        token: &str,
        timeout: Option<Duration>,
    ) -> LsFuture<'_, Result<DavLock, ()>> {
        let (path, token) = (path.clone(), token.to_string());
        Box::pin(async move {
            // Swept first: a client refreshing a lock it let expire must be
            // told, not silently handed the file back.
            self.sweep(&path).await;
            self.0.refresh(&path, &token, Self::capped(timeout)).await
        })
    }

    fn check(
        &self,
        path: &DavPath,
        principal: Option<&str>,
        ignore_principal: bool,
        deep: bool,
        submitted_tokens: &[String],
    ) -> LsFuture<'_, Result<(), DavLock>> {
        let (path, principal) = (path.clone(), principal.map(str::to_string));
        let tokens = submitted_tokens.to_vec();
        Box::pin(async move {
            self.sweep(&path).await;
            self.0
                .check(&path, principal.as_deref(), ignore_principal, deep, &tokens)
                .await
        })
    }

    fn discover(&self, path: &DavPath) -> LsFuture<'_, Vec<DavLock>> {
        let path = path.clone();
        Box::pin(async move {
            self.sweep(&path).await;
            self.0.discover(&path).await
        })
    }

    fn delete(&self, path: &DavPath) -> LsFuture<'_, Result<(), ()>> {
        let path = path.clone();
        Box::pin(async move { self.0.delete(&path).await })
    }
}

/// One mutex per path with a writer on it.
///
/// WebDAV locking does not cover this: a lock is only consulted for a client
/// that sends LOCK, and a plain PUT never does. Two concurrent PUTs otherwise
/// interleave into a byte-level splice of both bodies, with both clients told
/// 2xx. Serializing the write open makes the outcome last-writer-wins.
///
/// Keyed by the resolved absolute path, so two mounts onto the same file share
/// one mutex. The lock tree cannot do that: it keys on the URL, and the same
/// file has a different URL in a user mount and in a share.
static WRITE_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<tokio::sync::Mutex<()>>>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

fn write_lock(path: &Path) -> Arc<tokio::sync::Mutex<()>> {
    let mut map = WRITE_LOCKS.lock().unwrap_or_else(|e| e.into_inner());
    // Drop entries whose last writer finished, so the map holds in-flight
    // writes and not every file ever written.
    map.retain(|_, w| w.strong_count() > 0);
    if let Some(m) = map.get(path).and_then(Weak::upgrade) {
        return m;
    }
    let m = Arc::new(tokio::sync::Mutex::new(()));
    map.insert(path.to_path_buf(), Arc::downgrade(&m));
    m
}

// ---------------------------------------------------------------------------
// Mount: which roots a request sees, and where in the URL they live
// ---------------------------------------------------------------------------

/// The credentials `dav-server` carries through to [`FbFs`]: the roots this
/// request may touch, and how they are laid out under the mount point.
#[derive(Clone)]
pub struct Mount {
    /// URL segment → root. The segment is empty when `flat`.
    roots: Arc<Vec<(String, RootRow)>>,
    /// One root mounted directly at the mount point (a share), rather than as
    /// a child of a synthetic collection.
    flat: bool,
}

/// What a dav path addresses.
enum Target {
    /// The synthetic collection at the mount point that lists the roots.
    Roots,
    Item {
        root: RootRow,
        rel: String,
    },
}

/// The URL segment for each root: its display name, disambiguated with the
/// root id when two roots would otherwise claim the same one.
fn root_segments(state: &AppState, roots: Vec<RootRow>) -> Vec<(String, RootRow)> {
    let names: Vec<String> = roots.iter().map(|r| display_name(state, &r.path)).collect();
    roots
        .into_iter()
        .zip(&names)
        .map(|(r, name)| {
            let taken = names.iter().filter(|n| *n == name).count() > 1;
            let seg = match taken {
                true => format!("{name}-{}", r.id),
                false => name.clone(),
            };
            (seg, r)
        })
        .collect()
}

fn target(path: &DavPath, mount: &Mount) -> FsResult<Target> {
    let rel = path.as_rel_ospath();
    if mount.flat {
        let (_, root) = mount.roots.first().ok_or(FsError::NotFound)?;
        return Ok(Target::Item {
            root: root.clone(),
            rel: rel.to_string_lossy().into_owned(),
        });
    }
    let mut parts = rel.components();
    let Some(first) = parts.next() else {
        return Ok(Target::Roots);
    };
    let seg = first.as_os_str().to_string_lossy();
    let (_, root) = mount
        .roots
        .iter()
        .find(|(s, _)| s.as_str() == seg)
        .ok_or(FsError::NotFound)?;
    Ok(Target::Item {
        root: root.clone(),
        rel: parts.collect::<PathBuf>().to_string_lossy().into_owned(),
    })
}

/// [`target`], rejecting the synthetic collection.
fn item(path: &DavPath, mount: &Mount) -> FsResult<(RootRow, String)> {
    match target(path, mount)? {
        Target::Roots => Err(FsError::Forbidden),
        Target::Item { root, rel } => Ok((root, rel)),
    }
}

fn writable(root: &RootRow) -> FsResult<()> {
    match root.mode {
        Mode::Rw => Ok(()),
        Mode::Ro => Err(FsError::Forbidden),
    }
}

// ---------------------------------------------------------------------------
// The filesystem
// ---------------------------------------------------------------------------

#[derive(Clone)]
struct FbFs {
    state: Arc<AppState>,
}

impl GuardedFileSystem<Mount> for FbFs {
    fn open<'a>(
        &'a self,
        path: &'a DavPath,
        options: OpenOptions,
        mount: &'a Mount,
    ) -> FsFuture<'a, Box<dyn DavFile>> {
        Box::pin(async move {
            let (root, rel) = item(path, mount)?;
            if options.write || options.append || options.truncate || options.create {
                writable(&root)?;
            }
            let creating = options.create || options.create_new;
            let full = self
                .resolve(&root, rel, move |server_root, root_rel, rel| {
                    use crate::fs::FsError as E;
                    // Strict first, so a write lands on the file the path
                    // really names.
                    match crate::fs::resolve_path(server_root, root_rel, rel) {
                        Err(E::NotFound) if creating => {
                            let p = crate::fs::resolve_entry(server_root, root_rel, rel)?;
                            // Nothing resolved, yet the name is taken: a
                            // dangling symlink. Opening that with `create`
                            // would write wherever it points, which may be
                            // outside the root.
                            if std::fs::symlink_metadata(&p).is_ok() {
                                return Err(E::Forbidden);
                            }
                            Ok(p)
                        }
                        other => other,
                    }
                })
                .await?;
            // Taken before the open, so the truncate happens under it too,
            // and held until the `DavFile` is dropped, which is after the last
            // byte of the body has landed.
            let writing = options.write || options.append || options.truncate;
            let _write = match writing {
                true => Some(write_lock(&full).lock_owned().await),
                false => None,
            };
            let file = tokio::fs::OpenOptions::new()
                .read(options.read)
                .write(options.write)
                .append(options.append)
                .truncate(options.truncate)
                .create(options.create)
                .create_new(options.create_new)
                .open(&full)
                .await
                .map_err(|e| io_error(&e))?;
            Ok(Box::new(File { file, _write }) as Box<dyn DavFile>)
        })
    }

    /// `meta` decides whether a symlink is described as itself or as what it
    /// points at, and `dav-server` picks it per operation: `Data` for a
    /// listing, `DataSymlink` for the walk behind a recursive DELETE or COPY.
    /// Answering both with followed metadata makes a recursive DELETE descend
    /// into a linked directory and empty it.
    fn read_dir<'a>(
        &'a self,
        path: &'a DavPath,
        meta: ReadDirMeta,
        mount: &'a Mount,
    ) -> FsFuture<'a, FsStream<Box<dyn DavDirEntry>>> {
        Box::pin(async move {
            let listing = matches!(meta, ReadDirMeta::Data);
            let entries = match target(path, mount)? {
                Target::Roots => {
                    // That walk deletes the children before it asks to remove
                    // the collection, so refusing the mount point at
                    // `remove_dir` would come after every root was emptied.
                    // Refusing the listing stops it before anything is touched.
                    if !listing {
                        return Err(FsError::Forbidden);
                    }
                    self.root_entries(mount).await
                }
                Target::Item { root, rel } => {
                    // Same for a root's own top. It is a mount point, not a
                    // folder inside one. A whole root cannot be deleted, moved
                    // onto, or copied through the mount.
                    if rel.is_empty() && !listing {
                        return Err(FsError::Forbidden);
                    }
                    let full = self.resolve(&root, rel, crate::fs::resolve_path).await?;
                    // No `MAX_LIST_ENTRIES` cap here, on purpose. PROPFIND has
                    // no way to say "this listing was cut", so a sync client
                    // would read a truncated listing as "the rest was deleted"
                    // and mirror that.
                    blocking(move || {
                        let rd = std::fs::read_dir(&full).map_err(|e| io_error(&e))?;
                        Ok(rd
                            .flatten()
                            .filter_map(|e| {
                                // A link out of the root is still listed. It
                                // refuses to open.
                                let meta = match listing {
                                    true => match std::fs::metadata(e.path()) {
                                        Ok(m) => Meta::of(&m),
                                        // No target to stat: a dangling link.
                                        Err(_) => Meta::broken_link(
                                            &std::fs::symlink_metadata(e.path()).ok()?,
                                        ),
                                    },
                                    false => Meta::of(&std::fs::symlink_metadata(e.path()).ok()?),
                                };
                                Some(Entry {
                                    name: e.file_name().to_string_lossy().into_owned().into_bytes(),
                                    meta,
                                })
                            })
                            .collect())
                    })
                    .await?
                }
            };
            let stream = futures_util::stream::iter(
                entries
                    .into_iter()
                    .map(|e| Ok(Box::new(e) as Box<dyn DavDirEntry>)),
            );
            Ok(Box::pin(stream) as FsStream<Box<dyn DavDirEntry>>)
        })
    }

    fn metadata<'a>(
        &'a self,
        path: &'a DavPath,
        mount: &'a Mount,
    ) -> FsFuture<'a, Box<dyn DavMetaData>> {
        Box::pin(self.stat(path, mount, crate::fs::resolve_path, |p| {
            std::fs::metadata(p)
        }))
    }

    /// Metadata of the entry itself. `dav-server` asks this before a DELETE,
    /// a MOVE, and before overwriting a destination, precisely so it can act
    /// on a link rather than on what it names.
    fn symlink_metadata<'a>(
        &'a self,
        path: &'a DavPath,
        mount: &'a Mount,
    ) -> FsFuture<'a, Box<dyn DavMetaData>> {
        Box::pin(self.stat(path, mount, crate::fs::resolve_entry, |p| {
            std::fs::symlink_metadata(p)
        }))
    }

    fn create_dir<'a>(&'a self, path: &'a DavPath, mount: &'a Mount) -> FsFuture<'a, ()> {
        Box::pin(async move {
            let (root, rel) = item(path, mount)?;
            writable(&root)?;
            let (server_root, root_rel) = (self.state.root.clone(), root.path.clone());
            blocking(move || crate::fs::mkdir(&server_root, &root_rel, &rel).map_err(fs_error))
                .await
        })
    }

    /// Only ever called on an empty directory: `dav-server` walks a tree
    /// itself and removes the children first.
    fn remove_dir<'a>(&'a self, path: &'a DavPath, mount: &'a Mount) -> FsFuture<'a, ()> {
        Box::pin(async move {
            let (root, rel) = item(path, mount)?;
            writable(&root)?;
            let full = self.resolve(&root, rel, crate::fs::resolve_entry).await?;
            blocking(move || {
                // A symlink to a directory is listed as a collection, so this
                // is where DELETE lands on one. Unlink it rather than letting
                // `remove_dir` fail on a path that is not a directory.
                let meta = std::fs::symlink_metadata(&full).map_err(|e| io_error(&e))?;
                match meta.file_type().is_symlink() {
                    true => std::fs::remove_file(&full),
                    false => std::fs::remove_dir(&full),
                }
                .map_err(|e| io_error(&e))
            })
            .await
        })
    }

    fn remove_file<'a>(&'a self, path: &'a DavPath, mount: &'a Mount) -> FsFuture<'a, ()> {
        Box::pin(async move {
            let (root, rel) = item(path, mount)?;
            writable(&root)?;
            // Not followed: deleting a symlink removes the link, not the file
            // it names.
            let full = self.resolve(&root, rel, crate::fs::resolve_entry).await?;
            blocking(move || std::fs::remove_file(&full).map_err(|e| io_error(&e))).await
        })
    }

    fn rename<'a>(
        &'a self,
        from: &'a DavPath,
        to: &'a DavPath,
        mount: &'a Mount,
    ) -> FsFuture<'a, ()> {
        Box::pin(async move {
            let (src, dst) = (item(from, mount)?, item(to, mount)?);
            // A move takes the item out of the source root, so that root has
            // to be writable too.
            writable(&src.0)?;
            writable(&dst.0)?;
            let server_root = self.state.root.clone();
            blocking(move || {
                crate::fs::move_to(&server_root, &src.0.path, &src.1, &dst.0.path, &dst.1)
                    .map_err(fs_error)
            })
            .await
        })
    }

    /// Files only: `dav-server` walks a directory tree itself.
    fn copy<'a>(
        &'a self,
        from: &'a DavPath,
        to: &'a DavPath,
        mount: &'a Mount,
    ) -> FsFuture<'a, ()> {
        Box::pin(async move {
            let (src, dst) = (item(from, mount)?, item(to, mount)?);
            // Only the destination is written. Copying *out of* a read-only
            // root is fine, and is how a user gets a read-only folder's
            // contents into a writable one.
            writable(&dst.0)?;
            // The same mutex a PUT to this path would take, or a COPY and a
            // PUT racing for it interleave. Both resolve to the canonical
            // parent plus the name, so the keys agree.
            let full = self
                .resolve(&dst.0, dst.1.clone(), crate::fs::resolve_entry)
                .await?;
            let _write = write_lock(&full).lock_owned().await;
            let server_root = self.state.root.clone();
            blocking(move || {
                crate::fs::copy_file_to(&server_root, &src.0.path, &src.1, &dst.0.path, &dst.1)
                    .map_err(fs_error)
            })
            .await
        })
    }
}

impl FbFs {
    /// Metadata of `path`, resolved by `resolve` and read by `stat`.
    async fn stat(
        &self,
        path: &DavPath,
        mount: &Mount,
        resolve: fn(&Path, &str, &str) -> Result<PathBuf, crate::fs::FsError>,
        stat: fn(&Path) -> std::io::Result<std::fs::Metadata>,
    ) -> FsResult<Box<dyn DavMetaData>> {
        let (root, rel) = match target(path, mount)? {
            Target::Roots => return Ok(Box::new(Meta::synthetic_dir())),
            Target::Item { root, rel } => (root, rel),
        };
        let full = self.resolve(&root, rel, resolve).await?;
        let meta = blocking(move || stat(&full).map_err(|e| io_error(&e))).await?;
        Ok(Box::new(Meta::of(&meta)))
    }

    /// Run one of the [`crate::fs`] resolvers on the blocking pool.
    async fn resolve(
        &self,
        root: &RootRow,
        rel: String,
        f: impl FnOnce(&std::path::Path, &str, &str) -> Result<PathBuf, crate::fs::FsError>
        + Send
        + 'static,
    ) -> FsResult<PathBuf> {
        let (server_root, root_rel) = (self.state.root.clone(), root.path.clone());
        blocking(move || f(&server_root, &root_rel, &rel).map_err(fs_error)).await
    }

    /// The synthetic top-level listing: one entry per mounted root.
    async fn root_entries(&self, mount: &Mount) -> Vec<Entry> {
        let mut out = Vec::with_capacity(mount.roots.len());
        for (seg, root) in mount.roots.iter() {
            // A root that no longer resolves is skipped rather than reported
            // as broken: the JSON API hides it the same way.
            let (server_root, root_rel) = (self.state.root.clone(), root.path.clone());
            let Ok(full) = blocking(move || {
                crate::fs::resolve_root(&server_root, &root_rel).map_err(fs_error)
            })
            .await
            else {
                continue;
            };
            let meta = tokio::fs::metadata(&full)
                .await
                .map(|m| Meta::of(&m))
                .unwrap_or_else(|_| Meta::synthetic_dir());
            out.push(Entry {
                name: seg.clone().into_bytes(),
                meta,
            });
        }
        out
    }
}

// ---------------------------------------------------------------------------
// Filesystem value types
// ---------------------------------------------------------------------------

#[derive(Debug, Clone)]
struct Meta {
    len: u64,
    modified: SystemTime,
    is_dir: bool,
    is_symlink: bool,
}

impl Meta {
    fn of(m: &std::fs::Metadata) -> Self {
        Meta {
            len: m.len(),
            modified: m.modified().unwrap_or(UNIX_EPOCH),
            is_dir: m.is_dir(),
            is_symlink: m.file_type().is_symlink(),
        }
    }

    /// A listing entry whose target could not be stat'd: a dangling symlink.
    ///
    /// Described as an empty file, not as a link: `dav-server` drops any entry
    /// a listing reports as a symlink, and a sync client reads a file missing
    /// from PROPFIND as a deletion to mirror.
    fn broken_link(m: &std::fs::Metadata) -> Self {
        Meta {
            len: 0,
            is_dir: false,
            is_symlink: false,
            ..Meta::of(m)
        }
    }

    /// The mount point itself, which is not a directory on disk.
    fn synthetic_dir() -> Self {
        Meta {
            len: 0,
            modified: UNIX_EPOCH,
            is_dir: true,
            is_symlink: false,
        }
    }
}

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

    fn modified(&self) -> FsResult<SystemTime> {
        Ok(self.modified)
    }

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

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

#[derive(Debug)]
struct Entry {
    name: Vec<u8>,
    meta: Meta,
}

impl DavDirEntry for Entry {
    fn name(&self) -> Vec<u8> {
        self.name.clone()
    }

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

/// An open file. Plain async I/O: the path was already resolved and checked,
/// so nothing here needs the blocking pool.
#[derive(Debug)]
struct File {
    file: tokio::fs::File,
    /// Held for the life of a writable handle. See [`WRITE_LOCKS`].
    _write: Option<tokio::sync::OwnedMutexGuard<()>>,
}

/// Ceiling on one `read_bytes` allocation. `dav-server` asks for its own read
/// buffer size, but the count reaches us from the request, and a short read is
/// always a valid answer.
const MAX_READ: usize = 64 * 1024;

impl DavFile for File {
    fn metadata(&mut self) -> FsFuture<'_, Box<dyn DavMetaData>> {
        Box::pin(async move {
            let m = self.file.metadata().await.map_err(|e| io_error(&e))?;
            Ok(Box::new(Meta::of(&m)) as Box<dyn DavMetaData>)
        })
    }

    fn write_buf(&mut self, mut buf: Box<dyn Buf + Send>) -> FsFuture<'_, ()> {
        Box::pin(async move {
            self.file
                .write_all_buf(&mut buf)
                .await
                .map_err(|e| io_error(&e))
        })
    }

    fn write_bytes(&mut self, buf: Bytes) -> FsFuture<'_, ()> {
        Box::pin(async move { self.file.write_all(&buf).await.map_err(|e| io_error(&e)) })
    }

    fn read_bytes(&mut self, count: usize) -> FsFuture<'_, Bytes> {
        Box::pin(async move {
            let mut b = vec![0u8; count.min(MAX_READ)];
            let n = self.file.read(&mut b).await.map_err(|e| io_error(&e))?;
            b.truncate(n);
            Ok(Bytes::from(b))
        })
    }

    fn seek(&mut self, pos: SeekFrom) -> FsFuture<'_, u64> {
        Box::pin(async move { self.file.seek(pos).await.map_err(|e| io_error(&e)) })
    }

    fn flush(&mut self) -> FsFuture<'_, ()> {
        Box::pin(async move { self.file.flush().await.map_err(|e| io_error(&e)) })
    }
}

// ---------------------------------------------------------------------------
// Errors and blocking work
// ---------------------------------------------------------------------------

/// Run blocking filesystem work, mapping a panic or a shut-down runtime onto
/// a 500.
async fn blocking<T: Send + 'static>(
    f: impl FnOnce() -> FsResult<T> + Send + 'static,
) -> FsResult<T> {
    tokio::task::spawn_blocking(f)
        .await
        .map_err(|_| FsError::GeneralFailure)?
}

fn fs_error(e: crate::fs::FsError) -> FsError {
    use crate::fs::FsError as E;
    match e {
        E::NotFound | E::RootMissing => FsError::NotFound,
        E::Conflict => FsError::Exists,
        // `Invalid` is a rejected name, which is a refusal, not a 400 here:
        // WebDAV has no status for "that name is not allowed".
        E::NotADirectory | E::Forbidden | E::Invalid(_) => FsError::Forbidden,
    }
}

/// `dav-server` only derives this from `std::io::Error` when its own `localfs`
/// backend is compiled in, which it is not.
fn io_error(e: &std::io::Error) -> FsError {
    use std::io::ErrorKind as K;
    match e.kind() {
        K::NotFound => FsError::NotFound,
        K::PermissionDenied => FsError::Forbidden,
        K::AlreadyExists => FsError::Exists,
        K::CrossesDevices => FsError::IsRemote,
        // `read_dir` on a file. A refusal, not a server fault.
        K::NotADirectory => FsError::Forbidden,
        _ => FsError::GeneralFailure,
    }
}