// TRACES: FR-NC-13 | FR-NC-12
//! A library that is just a directory.
//!
//! The second [`RemoteBackend`], and the one that exists to prove the first
//! was an abstraction rather than a description. It serves a plain folder: a
//! local disk, an NFS or SMB mount, a Nextcloud desktop client's synced copy,
//! an external drive. No server, no account, no credential.
//!
//! # What it is honestly worse at, and why that is fine
//!
//! Nextcloud's fast path rests on directory ETags propagating up the tree, so
//! one request against the root proves a 50k-image library unchanged. A POSIX
//! directory's mtime says only that its own entry list changed — not that a
//! grandchild's *contents* did — so there is nothing here to propagate and
//! [`ChangeDetection::LocalEtags`] is the truthful answer. The engine reads
//! that and walks the tree every scan instead of pruning it.
//!
//! Which costs almost nothing, because the walk that was expensive was
//! expensive for a reason this backend does not have. Fifty thousand
//! `stat` calls against a local filesystem take well under a second; fifty
//! thousand `PROPFIND`s do not. The capability model is what lets both be
//! driven by the same engine at the speed each one actually runs at.
//!
//! # Identity
//!
//! [`RemoteId::Stable`] here is a hash of the path relative to the library
//! root. That gives the catalog what it needs — a `u64` that names a
//! photograph, is the same on every device looking at the same folder, and
//! does not change when the file is edited — which is what keys the thumbnail
//! shards and the face index (`catalog.md` §10.1).
//!
//! It does **not** survive a rename, and [`Capabilities::stable_ids`] says so.
//! A moved photograph is seen as a delete and an add, and its thumbnail is
//! derived again. That is the documented degradation for a backend without
//! server-assigned ids, and it is the right trade here: the alternative,
//! keying on the inode, is stable across a rename but *differs between
//! devices* and is reused by the filesystem after a delete — so two machines
//! would disagree about which photograph a thumbnail belonged to, and a
//! recycled inode would silently attach an old thumbnail to a new image.
//! Re-deriving a thumbnail is a cost; showing the wrong one is a bug.
//!
//! # Blocking
//!
//! Every filesystem call goes through the blocking pool. On a local disk that
//! is overkill; on the NFS mount this backend is most useful over, a stalled
//! server would otherwise wedge the async worker that made the call and every
//! other request sharing it.

use std::io::{Read, Seek, SeekFrom, Write};
use std::ops::Range;
use std::path::{Component, Path, PathBuf};

use std::sync::Arc;

use async_trait::async_trait;
use dr_sync::{
    Account, BackendProvider, Capabilities, ChangeDetection, Connection, Cursor, EntryKind,
    Materialisation, Precondition, RemoteBackend, RemoteChange, RemoteEntry, RemoteError, RemoteId,
    RemotePath, ServerPreviews, SignIn, Validator,
};

pub mod borrow;
pub mod vfs;

pub use borrow::{BorrowPool, BorrowStats, Borrowed};
pub use vfs::{NoVfs, Vfs};

/// The id written to [`Account::backend`] for a folder library.
///
/// On-disk configuration: changing it orphans every folder account.
pub const BACKEND_ID: &str = "folder";

/// TRACES: FR-NC-13 | FR-NC-6c
/// Registers the folder connector.
///
/// See [`dr_sync::provider`] for what each method is for.
///
/// # The detector
///
/// This crate knows how to read a directory and nothing about sync clients,
/// so the placeholder convention arrives from outside: whoever registers the
/// provider supplies a function that recognises a synced folder and returns
/// the [`Vfs`] for it. That keeps `dr-sync-folder` free of any client's
/// protocol, and it is what lets one connector serve a plain disk, a Nextcloud
/// tree, and whatever comes next.
///
/// Detection runs per connection because the answer changes: the same
/// directory offers hydration while the client is up and not while it is down.
/// Recognises a placeholder convention in a directory, if any applies.
///
/// Runs per connection rather than once, because the answer changes: the same
/// folder offers hydration while the sync client is up and not while it is
/// down.
pub type VfsDetector = dyn Fn(&Path) -> Option<Arc<dyn Vfs>> + Send + Sync;

#[derive(Default)]
pub struct FolderProvider {
    detect_vfs: Option<Box<VfsDetector>>,
}

impl FolderProvider {
    /// A folder connector that treats every directory as ordinary.
    pub fn new() -> Self {
        Self::default()
    }

    /// A folder connector that recognises placeholder conventions.
    pub fn with_vfs_detector(
        detect: impl Fn(&Path) -> Option<Arc<dyn Vfs>> + Send + Sync + 'static,
    ) -> Self {
        Self {
            detect_vfs: Some(Box::new(detect)),
        }
    }

    fn vfs_for(&self, root: &Path) -> Arc<dyn Vfs> {
        self.detect_vfs
            .as_ref()
            .and_then(|d| d(root))
            .unwrap_or_else(|| Arc::new(NoVfs))
    }
}

impl BackendProvider for FolderProvider {
    fn id(&self) -> &'static str {
        BACKEND_ID
    }

    fn display_name(&self) -> &'static str {
        "Folder"
    }

    fn endpoint_label(&self) -> &'static str {
        "Folder"
    }

    fn endpoint_placeholder(&self) -> &'static str {
        "/home/you/Pictures"
    }

    fn sign_in(&self) -> SignIn {
        SignIn::EndpointOnly
    }

    /// Check the directory before an account is written for it.
    ///
    /// A typo here would otherwise be stored, skip the launch screen on the
    /// next start, and surface as a scan that finds nothing — which reads as
    /// a broken library rather than a wrong path. The messages say what to fix.
    fn normalise_endpoint(&self, input: &str) -> Result<String, String> {
        let trimmed = input.trim();
        if trimmed.is_empty() {
            return Err("Choose the folder your photographs are in.".into());
        }

        // `~` is what a person types and what a shell would have expanded;
        // nothing expands it here, so a stored `~/Pictures` becomes a
        // directory literally named `~`.
        let expanded = match trimmed.strip_prefix("~/") {
            Some(rest) => match std::env::var_os("HOME") {
                Some(home) => PathBuf::from(home).join(rest),
                None => return Err("No home directory to expand ~ against.".into()),
            },
            None => PathBuf::from(trimmed),
        };

        if !expanded.is_absolute() {
            return Err("Give the full path to the folder, starting at /.".into());
        }
        if !expanded.exists() {
            return Err(format!("No folder at {}.", expanded.display()));
        }
        if !expanded.is_dir() {
            return Err(format!("{} is a file, not a folder.", expanded.display()));
        }

        // Resolved so a library reached through a symlink or a `..` is stored
        // under one name. Two spellings of one folder would otherwise be two
        // accounts with two catalogs indexing the same photographs.
        let canonical = expanded
            .canonicalize()
            .map_err(|e| format!("Cannot read {}: {e}", expanded.display()))?;

        Ok(canonical.to_string_lossy().into_owned())
    }

    fn account_for(&self, endpoint: &str) -> Result<Account, RemoteError> {
        Ok(Account::new(BACKEND_ID, endpoint))
    }

    fn connect(&self, conn: &Connection) -> Result<Box<dyn RemoteBackend>, RemoteError> {
        let root = Path::new(&conn.account.endpoint);
        Ok(Box::new(FolderBackend::with_vfs(root, self.vfs_for(root))?))
    }
}

/// TRACES: FR-NC-13 | FR-NC-4 | FR-NC-6c
/// A library rooted at a directory.
#[derive(Clone)]
pub struct FolderBackend {
    root: PathBuf,
    /// The placeholder convention in force, [`NoVfs`] for an ordinary folder.
    vfs: Arc<dyn Vfs>,
    caps: Capabilities,
}

impl std::fmt::Debug for FolderBackend {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FolderBackend")
            .field("root", &self.root)
            .field("vfs", &self.vfs.name())
            .finish_non_exhaustive()
    }
}

impl FolderBackend {
    /// Open the folder at `root`.
    ///
    /// The directory must exist now, and not existing is
    /// [`RemoteError::RootUnavailable`] — the library folder could not be
    /// opened, which is the whole of what this knows. A drive unplugged
    /// between sessions and a path typed wrongly at setup are the same
    /// observation from here, and both are answered the same way: keep the
    /// catalog, say which folder, and offer it again (FR-PLAT-AND-2).
    ///
    /// A mount dropped *during* a session surfaces per-operation as
    /// [`RemoteError::Network`] instead, which is what puts the app into
    /// offline mode and leaves the catalog readable, exactly as a dead server
    /// does.
    pub fn new(root: impl Into<PathBuf>) -> Result<Self, RemoteError> {
        Self::with_vfs(root, Arc::new(NoVfs))
    }

    /// Open the folder at `root` under a placeholder convention.
    ///
    /// The convention is chosen by the caller rather than sniffed here: the
    /// connector that knows how to talk to a given sync client is the one that
    /// knows whether it is running (see `dr_sync_nextcloud`).
    pub fn with_vfs(root: impl Into<PathBuf>, vfs: Arc<dyn Vfs>) -> Result<Self, RemoteError> {
        let root = root.into();
        if !root.is_dir() {
            // TRACES: FR-PLAT-AND-2
            // Not `Configuration`, which is where this lived while there was
            // nothing better. The distinction that matters is not "was the
            // account written wrongly" — which nothing here can know — but
            // "can this library be opened", and a caller that knows the
            // library was working yesterday can act on the second answer:
            // mark what it holds as offline rather than deleting it, and ask
            // for the folder again (FR-CAT-9).
            return Err(RemoteError::RootUnavailable(format!(
                "{} is not a folder",
                root.display()
            )));
        }
        // Reported per connection, not per backend: the same folder offers
        // hydration while the client is up and not while it is down, so this
        // cannot be a constant of the type (see `vfs`).
        let materialisation = if vfs.can_materialise() {
            Materialisation::OnDemand
        } else if vfs.name() == NoVfs.name() {
            Materialisation::Always
        } else {
            Materialisation::Placeholders
        };
        Ok(Self {
            root,
            vfs,
            caps: Capabilities {
                // A directory's mtime describes its own entry list and nothing
                // below it, so there is no propagation to exploit; the engine
                // walks and compares per entry.
                change_detection: ChangeDetection::LocalEtags,
                // A path hash does not survive a rename. See the module docs
                // for why the inode is not used instead.
                stable_ids: false,
                range_reads: true,
                // Not a protocol with a message size limit; a write is a write.
                chunked_upload: None,
                bulk_upload: false,
                conditional_write: true,
                server_previews: ServerPreviews::None,
                materialisation,
            },
        })
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    /// The local path for a remote path, refusing anything that escapes.
    ///
    /// The guard is not theoretical. A `RemotePath` is built from strings that
    /// reach us from a catalog written by another device and from filenames on
    /// the remote itself, and this backend resolves them against a real
    /// filesystem with the user's own permissions. `../../.ssh/id_ed25519` is
    /// a legal path segment; without this it would be a legal *read*.
    fn resolve(&self, path: &RemotePath) -> Result<PathBuf, RemoteError> {
        let rel = Path::new(path.as_str());
        for component in rel.components() {
            match component {
                Component::Normal(_) => {}
                Component::CurDir => {}
                Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
                    return Err(RemoteError::Configuration(format!(
                        "{path} leaves the library folder"
                    )));
                }
            }
        }
        Ok(self.root.join(rel))
    }

    /// Where a photograph's bytes are on disk, and whether they are really
    /// there.
    ///
    /// A placeholder lives under a *different* name — suffix-mode VFS renames
    /// on hydration rather than filling in place — so every read and write has
    /// to look for both. The materialised name is tried first: it is the
    /// common case, and the second `stat` is paid only when it misses.
    ///
    /// Returns the path to use and whether it holds real content.
    fn locate(&self, path: &RemotePath) -> Result<(PathBuf, bool), RemoteError> {
        let direct = self.resolve(path)?;
        if self.vfs.name() == NoVfs.name() || direct.exists() {
            return Ok((direct, true));
        }
        let stub = self.resolve(&RemotePath::new(
            self.vfs.placeholder_name(path.as_str()).into_owned(),
        ))?;
        if stub.exists() {
            return Ok((stub, false));
        }
        // Neither: genuinely missing. Report the name the caller asked for.
        Ok((direct, true))
    }

    /// The local path a [`RemoteId`] names.
    ///
    /// A stable id here is a hash and nothing can be resolved from it, exactly
    /// as a Nextcloud `oc:fileid` names no WebDAV endpoint. Callers hold the
    /// path alongside it in the catalog and pass that.
    fn resolve_id(&self, id: &RemoteId) -> Result<PathBuf, RemoteError> {
        match id {
            RemoteId::Path(p) => self.resolve(p),
            RemoteId::Stable(_) => Err(RemoteError::Unsupported(
                "a folder cannot be addressed by id; use RemoteId::Path",
            )),
        }
    }

    /// [`locate`](Self::locate) for an id.
    fn locate_id(&self, id: &RemoteId) -> Result<(PathBuf, bool), RemoteError> {
        match id {
            RemoteId::Path(p) => self.locate(p),
            RemoteId::Stable(_) => Err(RemoteError::Unsupported(
                "a folder cannot be addressed by id; use RemoteId::Path",
            )),
        }
    }
}

/// Run a filesystem operation off the async worker that asked for it.
///
/// See the module docs: a stalled network mount must not take the caller's
/// runtime with it.
async fn blocking<T, F>(f: F) -> Result<T, RemoteError>
where
    F: FnOnce() -> Result<T, RemoteError> + Send + 'static,
    T: Send + 'static,
{
    match tokio::task::spawn_blocking(f).await {
        Ok(r) => r,
        // The only way a blocking task fails to produce a result is a panic
        // inside it, which is a bug here rather than a condition the caller
        // can act on — but crashing the worker over it would lose a whole
        // scan, so it is reported like any other failure.
        Err(e) => Err(RemoteError::Protocol(format!("folder task failed: {e}"))),
    }
}

/// Map an IO failure to the error the engine already knows how to handle.
///
/// The classification is the point. [`RemoteError::indicates_offline`] drives
/// offline mode, so a vanished mount must reach it as `Network` — that is
/// precisely the "the library is unreachable, keep working from the catalog"
/// case — while a permissions problem must not, because going offline over one
/// forbidden file would hide a fixable problem behind a network banner.
fn map_io(e: std::io::Error, what: &str) -> RemoteError {
    use std::io::ErrorKind as K;
    match e.kind() {
        K::NotFound => RemoteError::NotFound(what.to_string()),
        K::PermissionDenied => RemoteError::PermissionDenied,
        K::AlreadyExists => RemoteError::PreconditionFailed,
        // ENOSPC and friends. Quota is what the engine calls "no room".
        K::StorageFull | K::QuotaExceeded | K::FileTooLarge => RemoteError::QuotaExceeded,
        // A dropped mount answers ESTALE/EIO/ENOTCONN, and the honest reading
        // is the same as a dead server: the library cannot be reached now, and
        // may be again shortly.
        K::HostUnreachable
        | K::NetworkUnreachable
        | K::NetworkDown
        | K::ConnectionAborted
        | K::ConnectionReset
        | K::NotConnected
        | K::BrokenPipe
        | K::TimedOut => RemoteError::Network(format!("{what}: {e}")),
        _ => RemoteError::Protocol(format!("{what}: {e}")),
    }
}

/// How long to wait for a requested download to land.
///
/// Generous, because the file may be tens of megabytes over a domestic
/// connection, and bounded, because a client that has stopped transferring
/// must not wedge a whole pass.
const MATERIALISE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);

/// How often to look for the materialised file while waiting.
const POLL: std::time::Duration = std::time::Duration::from_millis(200);

/// The identity of a file, from its path relative to the library root.
///
/// FNV-1a rather than `DefaultHasher`, whose output is explicitly unstable
/// between Rust releases: this value is written into the catalog and into the
/// thumbnail index, and must mean the same thing after a toolchain upgrade as
/// it did before one.
fn identity(path: &RemotePath) -> u64 {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for b in path.as_str().as_bytes() {
        h ^= *b as u64;
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    h
}

/// A file's validator: its size and modification time.
///
/// The pair, not either alone. An mtime with one-second granularity — which is
/// what some filesystems and most network mounts report — cannot distinguish
/// two writes in the same second, and a size alone cannot see an edit that
/// preserved it. Together they miss only a same-second write of identical
/// length, which for a photograph is a rewrite of the same frame.
fn validator_of(meta: &std::fs::Metadata) -> Validator {
    let (secs, nanos) = meta
        .modified()
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|d| (d.as_secs(), d.subsec_nanos()))
        .unwrap_or((0, 0));
    Validator::new(format!("{:x}-{:x}.{:x}", meta.len(), secs, nanos))
}

fn modified_secs(meta: &std::fs::Metadata) -> Option<i64> {
    meta.modified()
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|d| d.as_secs() as i64)
}

#[async_trait]
impl RemoteBackend for FolderBackend {
    fn capabilities(&self) -> &Capabilities {
        &self.caps
    }

    fn name(&self) -> &str {
        "Folder"
    }

    async fn list(
        &self,
        dir: &RemotePath,
        _since: Option<&Validator>,
    ) -> Result<Vec<RemoteEntry>, RemoteError> {
        let local = self.resolve(dir)?;
        let dir = dir.clone();
        let vfs = self.vfs.clone();
        blocking(move || {
            let read =
                std::fs::read_dir(&local).map_err(|e| map_io(e, &local.display().to_string()))?;

            let mut out = Vec::new();
            for entry in read {
                let entry = match entry {
                    Ok(e) => e,
                    // One unreadable entry must not fail the listing: a
                    // scan of a real library meets a broken symlink or a
                    // file being written, and abandoning the whole
                    // directory over it loses every photograph beside it.
                    Err(e) => {
                        log::debug!("skipping an entry in {}: {e}", local.display());
                        continue;
                    }
                };

                let name = entry.file_name();
                let Some(name) = name.to_str() else {
                    // A name that is not UTF-8 cannot round-trip through a
                    // `RemotePath`, and quietly mangling it would produce a
                    // path that addresses a different file — or none.
                    log::warn!("skipping a non-UTF-8 name in {}", local.display());
                    continue;
                };

                // `metadata`, not `symlink_metadata`: a symlinked shoot
                // folder is a normal way to assemble a library, and the
                // scan's depth limit is what stops a loop.
                let meta = match entry.metadata() {
                    Ok(m) => m,
                    Err(e) => {
                        log::debug!("skipping {name}: {e}");
                        continue;
                    }
                };

                // The photograph's own name, never the stub's. Identity is
                // derived from it, so downloading a file must not look like a
                // delete and an add — and `source_ref` must match what every
                // other device calls the same photograph.
                let stub = vfs.is_placeholder(name);
                let path = dir.join(vfs.real_name(name));

                out.push(RemoteEntry {
                    id: RemoteId::Stable(identity(&path)),
                    kind: if meta.is_dir() {
                        EntryKind::Directory
                    } else {
                        EntryKind::File
                    },
                    validator: validator_of(&meta),
                    // A stub is one byte and says nothing about what it stands
                    // for. Reporting that byte count would put a 1-byte
                    // `file_size` in the catalog for most of the library.
                    size: if stub { 0 } else { meta.len() },
                    modified: modified_secs(&meta),
                    // No renderer behind a folder; previews are extracted
                    // locally from the file itself.
                    has_preview: false,
                    materialised: !stub,
                    path,
                });
            }
            Ok(out)
        })
        .await
    }

    /// Not offered.
    ///
    /// A directory's mtime changes when its own entries are added or removed
    /// and at no other time, so it cannot answer the question this method
    /// exists for — "did anything below here change?". Returning it anyway
    /// would let a future caller prune a subtree whose contents had been
    /// edited, and hide those edits for as long as the folder list held still.
    async fn dir_validator(&self, _dir: &RemotePath) -> Result<Validator, RemoteError> {
        Err(RemoteError::Unsupported(
            "a folder's mtime does not propagate; use per-entry validators",
        ))
    }

    async fn delta(&self, _cursor: &Cursor) -> Result<(Vec<RemoteChange>, Cursor), RemoteError> {
        Err(RemoteError::Unsupported("a folder keeps no change feed"))
    }

    async fn get(&self, id: &RemoteId, range: Option<Range<u64>>) -> Result<Vec<u8>, RemoteError> {
        let (local, materialised) = self.locate_id(id)?;
        if !materialised {
            // The one byte in the stub is not the file. Returning it produced
            // a sidecar that parsed as empty and a thumbnail that never
            // decoded; reporting `NotFound` made the sidecar writer treat an
            // existing document as absent and overwrite it.
            return Err(RemoteError::NotMaterialised(local.display().to_string()));
        }
        blocking(move || {
            let what = local.display().to_string();
            let mut file = std::fs::File::open(&local).map_err(|e| map_io(e, &what))?;

            let Some(r) = range else {
                let mut buf = Vec::new();
                file.read_to_end(&mut buf).map_err(|e| map_io(e, &what))?;
                return Ok(buf);
            };

            // A short read at the end of the file is not an error: the header
            // extractor asks for a fixed window and the file may be smaller
            // than it, which is the ordinary case for a small JPEG.
            file.seek(SeekFrom::Start(r.start))
                .map_err(|e| map_io(e, &what))?;
            let want = r.end.saturating_sub(r.start);
            let mut buf = Vec::new();
            file.take(want)
                .read_to_end(&mut buf)
                .map_err(|e| map_io(e, &what))?;
            Ok(buf)
        })
        .await
    }

    async fn put(
        &self,
        path: &RemotePath,
        body: Vec<u8>,
        precond: Option<Precondition>,
    ) -> Result<Validator, RemoteError> {
        let (found, materialised) = self.locate(path)?;
        // Where the content belongs, which is not where a placeholder for it
        // sits — suffix-mode VFS gives the two different names.
        let local = self.resolve(path)?;

        // A stub is still this file, so what to do about it depends entirely
        // on what the caller is promising.
        let replaces = if materialised {
            None
        } else {
            match &precond {
                // Nothing here can satisfy it: the validator on a placeholder
                // describes the placeholder. The caller fetches the content
                // and tries again, which is what the typed error asks for.
                Some(Precondition::IfMatch(_)) => {
                    return Err(RemoteError::NotMaterialised(found.display().to_string()))
                }
                // Something *is* there — the file exists, only its content is
                // elsewhere — so a create-if-absent must fail.
                Some(Precondition::IfAbsent) => return Err(RemoteError::PreconditionFailed),
                // An unconditional write replaces the whole file, so there is
                // nothing in the stub worth reading and no reason to download
                // it first. Refusing here instead was a mistake: derived state
                // lives in the library folder and the client dehydrates it
                // like anything else, so a refusal meant sync could never
                // write to a folder it had been away from.
                None => Some(found),
            }
        };

        blocking(move || {
            let what = local.display().to_string();
            if let Some(parent) = local.parent() {
                std::fs::create_dir_all(parent)
                    .map_err(|e| map_io(e, &parent.display().to_string()))?;
            }

            match &precond {
                // Genuinely atomic: `O_CREAT | O_EXCL` is one syscall, so two
                // devices racing to create a sidecar cannot both win.
                Some(Precondition::IfAbsent) => {
                    let mut f = std::fs::OpenOptions::new()
                        .write(true)
                        .create_new(true)
                        .open(&local)
                        .map_err(|e| map_io(e, &what))?;
                    f.write_all(&body).map_err(|e| map_io(e, &what))?;
                    f.sync_all().map_err(|e| map_io(e, &what))?;
                    let meta = f.metadata().map_err(|e| map_io(e, &what))?;
                    return Ok(validator_of(&meta));
                }
                // Compare, then swap. A POSIX filesystem has no compare-and-
                // swap, so this narrows the window to the microseconds between
                // the `stat` and the `rename` rather than closing it. That is
                // still far tighter than the fallback the engine uses when a
                // backend declares no conditional write at all — comparing
                // revision counters *inside* the sidecar, which spans a whole
                // read-modify-write — which is why the capability is declared
                // rather than refused.
                Some(Precondition::IfMatch(expected)) => {
                    let meta = std::fs::metadata(&local).map_err(|e| map_io(e, &what))?;
                    if &validator_of(&meta) != expected {
                        return Err(RemoteError::PreconditionFailed);
                    }
                }
                None => {}
            }

            // Write beside the destination and rename over it, so a reader
            // never sees a half-written sidecar and an interrupted write
            // cannot destroy the file it was replacing. Beside, not in
            // `/tmp`: a rename across filesystems is not atomic, and on
            // Android `/tmp` is a different one.
            let tmp = local.with_extension(format!(
                "{}.darkroom-tmp",
                local.extension().and_then(|e| e.to_str()).unwrap_or("")
            ));
            let write = (|| -> Result<(), RemoteError> {
                let mut f = std::fs::File::create(&tmp).map_err(|e| map_io(e, &what))?;
                f.write_all(&body).map_err(|e| map_io(e, &what))?;
                f.sync_all().map_err(|e| map_io(e, &what))
            })();
            if let Err(e) = write {
                let _ = std::fs::remove_file(&tmp);
                return Err(e);
            }
            if let Err(e) = std::fs::rename(&tmp, &local) {
                let _ = std::fs::remove_file(&tmp);
                return Err(map_io(e, &what));
            }

            // The stub goes only once the content is safely in place. The
            // other order risks leaving neither, and in a synced tree an
            // absence is a deletion the client would propagate.
            if let Some(stub) = replaces {
                if let Err(e) = std::fs::remove_file(&stub) {
                    // The content landed, so the write succeeded; a leftover
                    // placeholder beside it is untidy rather than harmful, and
                    // the client reconciles the pair on its next pass.
                    log::warn!("removing placeholder {}: {e}", stub.display());
                }
            }

            let meta = std::fs::metadata(&local).map_err(|e| map_io(e, &what))?;
            Ok(validator_of(&meta))
        })
        .await
    }

    /// Delete a file, or an empty directory.
    ///
    /// **Not recursive, unlike WebDAV's `DELETE` on a collection.** The
    /// divergence is deliberate: a folder library is the user's own
    /// photographs on their own disk, with no server-side trash behind it, so
    /// a caller that passed the wrong path would have no way back. Nothing in
    /// the engine deletes a directory — the soft delete is a
    /// [`move_to`](RemoteBackend::move_to) into the trash folder — so refusing
    /// costs nothing and the guard is free.
    async fn delete(
        &self,
        id: &RemoteId,
        precond: Option<Precondition>,
    ) -> Result<(), RemoteError> {
        // Deliberately by whichever name is on disk: deleting a photograph
        // means deleting it whether or not its content happens to be here, and
        // a stub left behind would be re-listed by the next scan.
        let (local, _) = self.locate_id(id)?;
        blocking(move || {
            let what = local.display().to_string();
            let meta = std::fs::symlink_metadata(&local).map_err(|e| map_io(e, &what))?;

            match &precond {
                Some(Precondition::IfMatch(expected)) => {
                    if &validator_of(&meta) != expected {
                        return Err(RemoteError::PreconditionFailed);
                    }
                }
                // "Delete only if nothing is there" is not a thing to ask of a
                // delete; something is there or the `stat` above already
                // failed.
                Some(Precondition::IfAbsent) => {
                    return Err(RemoteError::Unsupported(
                        "IfAbsent is not meaningful on a delete",
                    ))
                }
                None => {}
            }

            if meta.is_dir() {
                std::fs::remove_dir(&local).map_err(|e| {
                    if e.kind() == std::io::ErrorKind::DirectoryNotEmpty {
                        RemoteError::Configuration(format!(
                            "{what} is not empty; a folder library will not delete a tree"
                        ))
                    } else {
                        map_io(e, &what)
                    }
                })
            } else {
                std::fs::remove_file(&local).map_err(|e| map_io(e, &what))
            }
        })
        .await
    }

    async fn move_to(&self, from: &RemoteId, to: &RemotePath) -> Result<(), RemoteError> {
        // Move whichever name exists. Trashing a photograph that is not
        // downloaded is a perfectly ordinary thing to do, and it must move the
        // stub — renaming a placeholder keeps it a placeholder.
        let (src, materialised) = self.locate_id(from)?;
        let dst = if materialised {
            self.resolve(to)?
        } else {
            // The destination keeps the placeholder suffix, or the client
            // would see a one-byte file appear where a photograph should be.
            self.resolve(&RemotePath::new(
                self.vfs.placeholder_name(to.as_str()).into_owned(),
            ))?
        };
        blocking(move || {
            let what = dst.display().to_string();
            // Parents first: the trash folder does not exist until the first
            // photograph is trashed, and the trait promises this creates it.
            if let Some(parent) = dst.parent() {
                std::fs::create_dir_all(parent)
                    .map_err(|e| map_io(e, &parent.display().to_string()))?;
            }

            match std::fs::rename(&src, &dst) {
                Ok(()) => Ok(()),
                // EXDEV. Both paths are inside one library root, so this
                // needs a root that spans a mount point — a shoot folder
                // that is its own mount, which is an ordinary way to attach
                // an archive drive. Copy and unlink rather than refusing:
                // the identity a rename would have preserved is a path hash
                // here, and it changes either way.
                Err(e) if e.raw_os_error() == Some(18) => {
                    std::fs::copy(&src, &dst).map_err(|e| map_io(e, &what))?;
                    std::fs::remove_file(&src).map_err(|e| {
                        // The copy landed. Leaving the original is a
                        // duplicate, which the next scan will show; losing
                        // the copy would be worse.
                        let _ = std::fs::remove_file(&dst);
                        map_io(e, &src.display().to_string())
                    })
                }
                Err(e) => Err(map_io(e, &what)),
            }
        })
        .await
    }

    /// TRACES: FR-NC-6c
    /// Ask the sync client to download a placeholder, and wait for it.
    ///
    /// Suffix-mode VFS *renames* on hydration, so completion is the
    /// materialised path appearing — not the stub changing size. Polling the
    /// original would wait forever.
    async fn materialise(&self, id: &RemoteId) -> Result<bool, RemoteError> {
        let (local, materialised) = self.locate_id(id)?;
        if materialised {
            // Already here. Not an error, and not a reason to ask again — and
            // `false` is what tells a borrower to leave it alone afterwards.
            return Ok(false);
        }
        let vfs = self.vfs.clone();
        let target = self.resolve_id(id)?;
        blocking(move || {
            vfs.materialise(&local)?;

            // The client acknowledges the command, not the transfer, so this
            // waits for the file to appear. A bounded wait: a hydration that
            // has not landed in this long is one the caller should be told
            // about rather than blocked on for ever — the pass can come back
            // to it.
            let deadline = std::time::Instant::now() + MATERIALISE_TIMEOUT;
            while std::time::Instant::now() < deadline {
                if target.is_file() {
                    return Ok(true);
                }
                std::thread::sleep(POLL);
            }
            Err(RemoteError::Network(format!(
                "{} did not download within {}s",
                target.display(),
                MATERIALISE_TIMEOUT.as_secs()
            )))
        })
        .await
    }

    /// TRACES: FR-NC-6c
    /// Hand the content back, leaving a placeholder.
    ///
    /// **Never a delete.** In a synced tree removing the file propagates the
    /// removal to the server; the client is asked to dehydrate, and if it
    /// cannot the content simply stays.
    async fn dematerialise(&self, id: &RemoteId) -> Result<(), RemoteError> {
        let (local, materialised) = self.locate_id(id)?;
        if !materialised {
            return Ok(());
        }
        let vfs = self.vfs.clone();
        blocking(move || vfs.dematerialise(&local)).await
    }

    async fn create_dir(&self, path: &RemotePath) -> Result<(), RemoteError> {
        let local = self.resolve(path)?;
        blocking(move || {
            // `create_dir_all` makes parents and succeeds on one that already
            // exists, which is exactly the contract.
            std::fs::create_dir_all(&local).map_err(|e| map_io(e, &local.display().to_string()))
        })
        .await
    }
}

#[cfg(test)]
mod tests;