//! The mount on macOS: an NFSv3 server on localhost (`nfsserve`) that macOS's
//! own NFS client mounts with `mount_nfs`, so nothing is installed and
//! nothing needs root. It serves the same overlay FUSE serves on Linux.
//!
//! A port on localhost is open to every user of the machine, and a FUSE mount
//! answers only the user who made it, so this one does the same:
//! - The mount request is answered once, for an export path with a secret in
//!   it, and every file handle starts with another secret, so nobody else can
//!   mount it or make up a handle.
//! - The port the system mounts is a relay in front of the server that passes
//!   on only calls made as this user (or root), so other users can't go
//!   through the mounted folder either.
//!
//! macOS keeps files of its own next to people's (`._name` for extended
//! attributes, `.nfs.*` for a file deleted while open, Finder's `.DS_Store`):
//! they're kept, but out of listings, so saves never see them.
//!
//! The serving process holds `mount.lock` in the workspace's backing, so a
//! mount whose process is gone (a crash) is told apart and taken down before
//! the next one starts. It stops when its mount goes (`on mount --stop`).

use std::collections::{HashMap, HashSet};
use std::ffi::{CStr, OsStr};
use std::fs;
use std::ops::Bound;
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, UNIX_EPOCH};

use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use nfsserve::nfs::{
    FSF_CANSETTIME, FSF_HOMOGENEOUS, FSF_SYMLINK, fattr3, fileid3, filename3, fsinfo3, ftype3, nfs_fh3, nfspath3, nfsstat3, nfstime3, post_op_attr, sattr3, set_atime, set_mode3,
    set_mtime, set_size3, specdata3,
};
use nfsserve::tcp::{NFSTcp, NFSTcpListener};
use nfsserve::vfs::{DirEntry, NFSFileSystem, ReadDirResult, VFSCapabilities};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

use super::overlay::{Attr, Errno, Kind, Node, Overlay, SetTime, join, parent, set_times};
use crate::app::App;
use crate::ws;

/// The options `mount_nfs` mounts with. The server also answers the mount
/// protocol on its port, so no portmapper is asked. `soft,intr`: if this
/// process dies, calls fail instead of hanging. Locks stay on this machine,
/// where everything using the folder is.
fn mount_options(port: u16) -> String {
    format!("vers=3,tcp,port={port},mountport={port},soft,intr,locallocks,noquota,nobrowse,rsize=1048576,wsize=1048576,actimeo=1")
}

/// How often the server checks that its mount is still there.
const MOUNT_CHECK: Duration = Duration::from_millis(500);

/// The largest RPC message the relay passes on: a 1 MiB write and its header.
const MAX_MESSAGE: usize = 4 << 20;

/// Every mount on this machine: (where, what), from the kernel's table
/// without asking any filesystem (a dead server's mount doesn't hang this).
fn mounts() -> Vec<(PathBuf, String)> {
    // SAFETY: a null buffer asks only how many there are.
    let n = unsafe { libc::getfsstat(std::ptr::null_mut(), 0, libc::MNT_NOWAIT) };
    if n <= 0 {
        return Vec::new();
    }
    // Room for a few more, in case one comes meanwhile.
    let mut all: Vec<libc::statfs> = vec![unsafe { std::mem::zeroed() }; n as usize + 8];
    let size = (all.len() * std::mem::size_of::<libc::statfs>()) as libc::c_int;
    // SAFETY: a buffer of `size` bytes of statfs, as getfsstat takes.
    let n = unsafe { libc::getfsstat(all.as_mut_ptr(), size, libc::MNT_NOWAIT) };
    all.truncate(n.max(0) as usize);
    all.iter()
        .map(|s| {
            // SAFETY: the kernel NUL-terminates both names.
            let on = unsafe { CStr::from_ptr(s.f_mntonname.as_ptr()) };
            let from = unsafe { CStr::from_ptr(s.f_mntfromname.as_ptr()) };
            (PathBuf::from(OsStr::from_bytes(on.to_bytes())), from.to_string_lossy().into_owned())
        })
        .collect()
}

/// What's mounted at `root` (the topmost mount there), if anything.
fn mounted_from(root: &Path) -> Option<String> {
    mounts().into_iter().rev().find(|(on, _)| on == root).map(|(_, from)| from)
}

/// Whether something is mounted at `root`.
pub fn mounted(root: &Path) -> bool {
    mounted_from(root).is_some()
}

/// The lock the serving process holds.
fn lock_path(backing: &Path) -> PathBuf {
    backing.join(ws::DIR).join("mount.lock")
}

/// Whether a process is serving the workspace whose backing is `backing`.
fn serving(backing: &Path) -> bool {
    use fs4::fs_std::FileExt;
    let Ok(file) = fs::OpenOptions::new().create(true).truncate(false).write(true).open(lock_path(backing)) else { return false };
    match file.try_lock_exclusive() {
        Ok(true) => {
            let _ = FileExt::unlock(&file);
            false
        }
        _ => true,
    }
}

/// Takes down a mount at `root` whose process is gone, so a new one can start.
pub fn clear_dead(root: &Path, backing: &Path) -> Result<()> {
    if mounted(root) && !serving(backing) {
        force_unmount(root)?;
    }
    Ok(())
}

fn run_quietly(program: &str, args: &[&str], root: &Path) -> bool {
    Command::new(program).args(args).arg(root).stdout(std::process::Stdio::null()).stderr(std::process::Stdio::null()).status().is_ok_and(|s| s.success())
}

fn force_unmount(root: &Path) -> Result<()> {
    if run_quietly("/sbin/umount", &["-f"], root) || run_quietly("/usr/sbin/diskutil", &["unmount", "force"], root) || !mounted(root) {
        return Ok(());
    }
    Err(anyhow!("couldn't take down the stopped mount at {}", root.display()))
}

/// The backing of the workspace a mount serves, from what's mounted
/// (`127.0.0.1:/onset/<workspace>/<secret>`).
fn backing_of(from: &str) -> Option<PathBuf> {
    let (_, export) = from.split_once(":/onset/")?;
    let (workspace, _) = export.split_once('/')?;
    Some(super::backing(workspace))
}

/// Stops the mount at `root`, and waits for its process to let go.
pub fn unmount(root: &Path) -> Result<()> {
    let Some(from) = mounted_from(root) else { return Ok(()) };
    let backing = backing_of(&from);
    if backing.as_deref().is_some_and(|b| !serving(b)) {
        return force_unmount(root);
    }
    if !(run_quietly("/sbin/umount", &[], root) || run_quietly("/usr/sbin/diskutil", &["unmount"], root)) && mounted(root) {
        return Err(anyhow!("couldn't unmount {} (is something still using it?)", root.display()));
    }
    let deadline = Instant::now() + Duration::from_secs(5);
    while backing.as_deref().is_some_and(serving) && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(20));
    }
    Ok(())
}

/// Why the mount didn't start, for `on get --lazy` and `on mount`: what the
/// serving process said, and that a full workspace always works.
pub fn stopped(root: &Path, said: &str) -> anyhow::Error {
    let said: Vec<&str> = said.lines().map(|l| l.strip_prefix("error: ").unwrap_or(l)).filter(|l| !l.trim().is_empty() && !l.starts_with("next:")).collect();
    crate::out::Fail::new("failed", format!("the lazy mount at {} didn't start: {}", root.display(), said.join("; ")))
        .state("the workspace is kept, without its files")
        .next("on get <project> <folder> without --lazy gets every file now")
        .into_anyhow()
}

/// `on mount --serve <folder>`: serves the mount until it's unmounted.
pub fn serve(app: &App, root: &Path) -> Result<()> {
    // Nothing of this process stays in the folder it covers.
    let _ = std::env::set_current_dir("/");
    let overlay = Overlay::load(app, root)?;
    let lock = fs::OpenOptions::new().create(true).truncate(false).write(true).open(lock_path(&overlay.backing))?;
    {
        use fs4::fs_std::FileExt;
        // Another process serving it (two commands starting it at once, or one stopping): wait for it.
        let deadline = Instant::now() + Duration::from_secs(10);
        while !matches!(lock.try_lock_exclusive(), Ok(true)) {
            if mounted(root) {
                return Ok(());
            }
            if Instant::now() > deadline {
                return Err(anyhow!("another process is serving {}", root.display()));
            }
            std::thread::sleep(Duration::from_millis(50));
        }
    }
    if mounted(root) {
        return Ok(());
    }
    let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build()?;
    let result = runtime.block_on(run(overlay, root));
    runtime.shutdown_background();
    drop(lock);
    result
}

async fn run(overlay: Overlay, root: &Path) -> Result<()> {
    // The workspace (so `on mount --stop` finds its process), and a secret.
    let workspace = overlay.backing.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
    let export = format!("onset/{workspace}/{}", hex(&rand::random::<[u8; 16]>()));
    let fs = Lazy { overlay, secret: rand::random(), mounted: AtomicBool::new(false), listed: Mutex::new(HashMap::new()) };
    let mut server = NFSTcpListener::bind("127.0.0.1:0", fs).await.context("starting the NFS server")?;
    server.with_export_name(&export);
    let inner = server.get_listen_port();
    tokio::spawn(async move { server.handle_forever().await });
    let front = TcpListener::bind("127.0.0.1:0").await.context("starting the NFS server")?;
    let port = front.local_addr()?.port();
    // SAFETY: getuid can't fail.
    tokio::spawn(relay(front, inner, unsafe { libc::getuid() }));

    let (options, source, point) = (mount_options(port), format!("127.0.0.1:/{export}"), root.to_path_buf());
    let out = tokio::task::spawn_blocking(move || Command::new("/sbin/mount_nfs").arg("-o").arg(options).arg(source).arg(point).output()).await?.context("running /sbin/mount_nfs")?;
    if !out.status.success() {
        let said = String::from_utf8_lossy(&out.stderr).trim().replace('\n', " ");
        return Err(anyhow!("macOS's NFS client (mount_nfs) couldn't mount it ({}): {said}", out.status));
    }
    eprintln!("on mount: serving {} ({}) on 127.0.0.1:{port}", root.display(), mounted_from(root).unwrap_or_default());
    // Until it's unmounted (or something else is mounted over it).
    let mut check = tokio::time::interval(MOUNT_CHECK);
    while mounted_from(root).is_some_and(|from| from.contains(&export)) {
        check.tick().await;
    }
    Ok(())
}

fn hex(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

/// Files macOS keeps for itself next to people's: left out of listings
/// unless the tree has them.
fn bookkeeping(name: &str) -> bool {
    name.starts_with("._") || name.starts_with(".nfs.") || name == ".DS_Store"
}

struct Lazy {
    overlay: Overlay,
    /// Starts every file handle, so handles can't be made up.
    secret: [u8; 16],
    /// Whether the mount request was answered (it's answered once).
    mounted: AtomicBool,
    /// By folder, the name each entry was listed under: a listing's next
    /// page starts after it, even if the entry moved since.
    listed: Mutex<HashMap<u64, HashMap<u64, String>>>,
}

fn status(e: Errno) -> nfsstat3 {
    match e.0 {
        libc::ENOENT => nfsstat3::NFS3ERR_NOENT,
        libc::EEXIST => nfsstat3::NFS3ERR_EXIST,
        libc::ENOTDIR => nfsstat3::NFS3ERR_NOTDIR,
        libc::EISDIR => nfsstat3::NFS3ERR_ISDIR,
        libc::ENOTEMPTY => nfsstat3::NFS3ERR_NOTEMPTY,
        libc::EINVAL => nfsstat3::NFS3ERR_INVAL,
        libc::EACCES => nfsstat3::NFS3ERR_ACCES,
        libc::EPERM => nfsstat3::NFS3ERR_PERM,
        libc::ENOSPC => nfsstat3::NFS3ERR_NOSPC,
        libc::EDQUOT => nfsstat3::NFS3ERR_DQUOT,
        libc::EROFS => nfsstat3::NFS3ERR_ROFS,
        libc::EFBIG => nfsstat3::NFS3ERR_FBIG,
        libc::ENAMETOOLONG => nfsstat3::NFS3ERR_NAMETOOLONG,
        libc::EXDEV => nfsstat3::NFS3ERR_XDEV,
        libc::EMLINK => nfsstat3::NFS3ERR_MLINK,
        _ => nfsstat3::NFS3ERR_IO,
    }
}

fn nfs_time(ns: i64) -> nfstime3 {
    let ns = ns.max(0);
    nfstime3 { seconds: (ns / 1_000_000_000) as u32, nseconds: (ns % 1_000_000_000) as u32 }
}

fn client_time(t: &nfstime3) -> SetTime {
    SetTime::At(UNIX_EPOCH + Duration::new(t.seconds.into(), t.nseconds))
}

fn atime(t: &set_atime) -> Option<SetTime> {
    match t {
        set_atime::DONT_CHANGE => None,
        set_atime::SET_TO_SERVER_TIME => Some(SetTime::Now),
        set_atime::SET_TO_CLIENT_TIME(t) => Some(client_time(t)),
    }
}

fn mtime(t: &set_mtime) -> Option<SetTime> {
    match t {
        set_mtime::DONT_CHANGE => None,
        set_mtime::SET_TO_SERVER_TIME => Some(SetTime::Now),
        set_mtime::SET_TO_CLIENT_TIME(t) => Some(client_time(t)),
    }
}

fn fattr(a: &Attr) -> fattr3 {
    fattr3 {
        ftype: match a.kind {
            Kind::Dir => ftype3::NF3DIR,
            Kind::File => ftype3::NF3REG,
            Kind::Link => ftype3::NF3LNK,
        },
        mode: a.perm.into(),
        nlink: a.nlink,
        uid: a.uid,
        gid: a.gid,
        size: a.size,
        used: a.blocks * 512,
        rdev: specdata3::default(),
        fsid: 0,
        fileid: a.ino,
        atime: nfs_time(a.atime_ns),
        mtime: nfs_time(a.mtime_ns),
        ctime: nfs_time(a.ctime_ns),
    }
}

fn name_of(name: &filename3) -> &OsStr {
    OsStr::from_bytes(&name.0)
}

/// Runs blocking work (the disk, a fetch from the host) without holding up
/// the server's other calls.
fn blocking<T>(f: impl FnOnce() -> T) -> T {
    tokio::task::block_in_place(f)
}

impl Lazy {
    /// What `path` shows, or STALE if it's gone (a handle the client kept).
    fn attr_at(&self, path: &str) -> Result<fattr3, nfsstat3> {
        let mut st = self.overlay.state.lock().unwrap();
        let node = self.overlay.find(&st, path).ok_or(nfsstat3::NFS3ERR_STALE)?;
        Ok(fattr(&self.overlay.attr(&mut st, path, &node)))
    }

    fn path(&self, id: fileid3) -> Result<String, nfsstat3> {
        self.overlay.path_of(&self.overlay.state.lock().unwrap(), id).map_err(|_| nfsstat3::NFS3ERR_STALE)
    }

    fn child(&self, dir: fileid3, name: &filename3) -> Result<String, nfsstat3> {
        self.overlay.child(&self.overlay.state.lock().unwrap(), dir, name_of(name)).map_err(status)
    }

    /// The ID and attributes of `path`, just made.
    fn made(&self, path: &str) -> Result<(fileid3, fattr3), nfsstat3> {
        let attr = self.attr_at(path).map_err(|_| nfsstat3::NFS3ERR_IO)?;
        Ok((attr.fileid, attr))
    }

    /// Fetches a file of the tree from the host into the store before the
    /// overlay is locked, so other calls aren't held up meanwhile (copying
    /// it up is quick after).
    fn prefetch(&self, path: &str) {
        let node = self.overlay.find(&self.overlay.state.lock().unwrap(), path);
        if let Some(Node::File(e, _)) = node {
            let _ = self.overlay.content(path, &e);
        }
    }

    /// Whether listings show `name` in folder `dir`.
    fn shows(&self, dir: &str, name: &str) -> bool {
        !bookkeeping(name) || self.overlay.in_tree(&join(dir, name))
    }
}

#[async_trait]
impl NFSFileSystem for Lazy {
    fn capabilities(&self) -> VFSCapabilities {
        VFSCapabilities::ReadWrite
    }

    fn root_dir(&self) -> fileid3 {
        1
    }

    fn id_to_fh(&self, id: fileid3) -> nfs_fh3 {
        nfs_fh3 { data: [&self.secret[..], &id.to_le_bytes()].concat() }
    }

    fn fh_to_id(&self, fh: &nfs_fh3) -> Result<fileid3, nfsstat3> {
        match fh.data.split_at_checked(self.secret.len()) {
            Some((secret, id)) if secret == self.secret => id.try_into().map(u64::from_le_bytes).map_err(|_| nfsstat3::NFS3ERR_BADHANDLE),
            _ => Err(nfsstat3::NFS3ERR_BADHANDLE),
        }
    }

    /// The mount request (the server has checked the export path): the root,
    /// once, for the `mount_nfs` this process ran.
    async fn path_to_id(&self, path: &[u8]) -> Result<fileid3, nfsstat3> {
        if path == b"/" && !self.mounted.swap(true, Ordering::SeqCst) { Ok(1) } else { Err(nfsstat3::NFS3ERR_ACCES) }
    }

    async fn lookup(&self, dirid: fileid3, filename: &filename3) -> Result<fileid3, nfsstat3> {
        blocking(|| {
            let mut st = self.overlay.state.lock().unwrap();
            let dir = self.overlay.path_of(&st, dirid).map_err(|_| nfsstat3::NFS3ERR_STALE)?;
            match &filename.0[..] {
                b"." => return Ok(dirid),
                b".." => return Ok(if dir.is_empty() { 1 } else { self.overlay.ino_of(&mut st, parent(&dir)) }),
                _ => {}
            }
            let path = self.overlay.child(&st, dirid, name_of(filename)).map_err(|_| nfsstat3::NFS3ERR_NOENT)?;
            match self.overlay.find(&st, &path) {
                Some(_) => Ok(self.overlay.ino_of(&mut st, &path)),
                None => Err(nfsstat3::NFS3ERR_NOENT),
            }
        })
    }

    async fn getattr(&self, id: fileid3) -> Result<fattr3, nfsstat3> {
        blocking(|| self.attr_at(&self.path(id)?))
    }

    async fn setattr(&self, id: fileid3, attr: sattr3) -> Result<fattr3, nfsstat3> {
        blocking(|| {
            let path = self.path(id)?;
            let mode = match attr.mode {
                set_mode3::mode(m) => Some(m),
                set_mode3::Void => None,
            };
            let size = match attr.size {
                set_size3::size(n) => Some(n),
                set_size3::Void => None,
            };
            // Owners stay as they are: the mount is one user's.
            self.prefetch(&path);
            let st = self.overlay.state.lock().unwrap();
            self.overlay.setattr(&st, &path, mode, size, None, atime(&attr.atime), mtime(&attr.mtime)).map_err(status)?;
            drop(st);
            self.attr_at(&path)
        })
    }

    async fn read(&self, id: fileid3, offset: u64, count: u32) -> Result<(Vec<u8>, bool), nfsstat3> {
        blocking(|| {
            let path = self.path(id)?;
            self.prefetch(&path);
            let file = {
                let st = self.overlay.state.lock().unwrap();
                self.overlay.open(&st, &path, true, false, false).map_err(status)?
            };
            let len = file.metadata().map_err(|e| status(Errno::from(e)))?.len();
            let mut buf = vec![0u8; (count as u64).min(len.saturating_sub(offset)) as usize];
            let mut got = 0;
            while got < buf.len() {
                match file.read_at(&mut buf[got..], offset + got as u64) {
                    Ok(0) => break,
                    Ok(n) => got += n,
                    Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
                    Err(e) => return Err(status(Errno::from(e))),
                }
            }
            buf.truncate(got);
            Ok((buf, offset + got as u64 >= len))
        })
    }

    async fn write(&self, id: fileid3, offset: u64, data: &[u8]) -> Result<fattr3, nfsstat3> {
        blocking(|| {
            let path = self.path(id)?;
            self.prefetch(&path);
            let file = {
                let st = self.overlay.state.lock().unwrap();
                self.overlay.open(&st, &path, false, true, false).map_err(status)?
            };
            file.write_all_at(data, offset).map_err(|e| status(Errno::from(e)))?;
            self.attr_at(&path)
        })
    }

    async fn create(&self, dirid: fileid3, filename: &filename3, attr: sattr3) -> Result<(fileid3, fattr3), nfsstat3> {
        blocking(|| {
            let path = self.child(dirid, filename)?;
            let mode = match attr.mode {
                set_mode3::mode(m) => m,
                set_mode3::Void => 0o644,
            };
            let st = self.overlay.state.lock().unwrap();
            self.overlay.create(&st, &path, mode, false, matches!(attr.size, set_size3::size(0))).map_err(status)?;
            let (at, mt) = (atime(&attr.atime), mtime(&attr.mtime));
            if at.is_some() || mt.is_some() {
                self.overlay.setattr(&st, &path, None, None, None, at, mt).map_err(status)?;
            }
            drop(st);
            self.made(&path)
        })
    }

    async fn create_exclusive(&self, dirid: fileid3, filename: &filename3) -> Result<fileid3, nfsstat3> {
        blocking(|| {
            let path = self.child(dirid, filename)?;
            let st = self.overlay.state.lock().unwrap();
            self.overlay.create(&st, &path, 0o644, true, false).map_err(status)?;
            drop(st);
            Ok(self.made(&path)?.0)
        })
    }

    async fn mkdir(&self, dirid: fileid3, dirname: &filename3) -> Result<(fileid3, fattr3), nfsstat3> {
        blocking(|| {
            let path = self.child(dirid, dirname)?;
            let st = self.overlay.state.lock().unwrap();
            self.overlay.mkdir(&st, &path, 0o755).map_err(status)?;
            drop(st);
            self.made(&path)
        })
    }

    /// REMOVE and RMDIR both come here.
    async fn remove(&self, dirid: fileid3, filename: &filename3) -> Result<(), nfsstat3> {
        blocking(|| {
            let path = self.child(dirid, filename)?;
            let mut st = self.overlay.state.lock().unwrap();
            let node = self.overlay.find(&st, &path).ok_or(nfsstat3::NFS3ERR_NOENT)?;
            if node.is_dir() {
                // macOS's own files are all that's left: they go with it.
                let left = self.overlay.listing(&st, &path);
                if !left.is_empty() && left.iter().all(|(name, _)| !self.shows(&path, name)) {
                    for (name, _) in left {
                        let _ = fs::remove_file(self.overlay.real(&join(&path, &name)));
                    }
                }
                self.overlay.rmdir(&mut st, path.clone()).map_err(status)?;
            } else {
                self.overlay.unlink(&mut st, path.clone()).map_err(status)?;
            }
            // The folder shows a new time, so the client reads it again.
            let up = parent(&path);
            if self.overlay.real_dir(up).is_ok() {
                let _ = set_times(&self.overlay.real(up), None, Some(SetTime::Now));
            }
            Ok(())
        })
    }

    async fn rename(&self, from_dirid: fileid3, from_filename: &filename3, to_dirid: fileid3, to_filename: &filename3) -> Result<(), nfsstat3> {
        blocking(|| {
            let from = self.child(from_dirid, from_filename)?;
            let to = self.child(to_dirid, to_filename)?;
            let mut st = self.overlay.state.lock().unwrap();
            self.overlay.rename(&mut st, from, to, false).map_err(status)
        })
    }

    /// A page of folder `dirid`, in name order, after the entry the last page
    /// ended on (`start_after`, 0 for the first page).
    async fn readdir(&self, dirid: fileid3, start_after: fileid3, max_entries: usize) -> Result<ReadDirResult, nfsstat3> {
        blocking(|| {
            let mut st = self.overlay.state.lock().unwrap();
            let path = self.overlay.path_of(&st, dirid).map_err(|_| nfsstat3::NFS3ERR_STALE)?;
            if !self.overlay.find(&st, &path).is_some_and(|n| n.is_dir()) {
                return Err(nfsstat3::NFS3ERR_NOTDIR);
            }
            let names = self.overlay.names(&path);
            let mut listed = self.listed.lock().unwrap();
            let seen = listed.entry(dirid).or_default();
            let after = if start_after == 0 {
                Bound::Unbounded
            } else {
                // The name it was listed under, or else the one it has now if it's still in this folder.
                let now = self.overlay.path_of(&st, start_after).ok().filter(|p| !p.is_empty() && parent(p) == path).map(|p| p.rsplit('/').next().unwrap_or_default().to_string());
                Bound::Excluded(seen.get(&start_after).cloned().or(now).ok_or(nfsstat3::NFS3ERR_BAD_COOKIE)?)
            };
            let (mut entries, mut end) = (Vec::new(), true);
            for name in names.range((after, Bound::Unbounded)) {
                if !self.shows(&path, name) {
                    continue;
                }
                let child = join(&path, name);
                let Some(node) = self.overlay.find(&st, &child) else { continue };
                if entries.len() >= max_entries.max(1) {
                    end = false;
                    break;
                }
                let attr = self.overlay.attr(&mut st, &child, &node);
                seen.insert(attr.ino, name.clone());
                entries.push(DirEntry { fileid: attr.ino, name: name.as_bytes().into(), attr: fattr(&attr) });
            }
            Ok(ReadDirResult { entries, end })
        })
    }

    async fn symlink(&self, dirid: fileid3, linkname: &filename3, symlink: &nfspath3, _attr: &sattr3) -> Result<(fileid3, fattr3), nfsstat3> {
        blocking(|| {
            let path = self.child(dirid, linkname)?;
            let st = self.overlay.state.lock().unwrap();
            self.overlay.symlink(&st, &path, Path::new(OsStr::from_bytes(&symlink.0))).map_err(status)?;
            drop(st);
            self.made(&path)
        })
    }

    async fn readlink(&self, id: fileid3) -> Result<nfspath3, nfsstat3> {
        blocking(|| {
            let path = self.path(id)?;
            let st = self.overlay.state.lock().unwrap();
            self.overlay.readlink(&st, &path).map(nfspath3::from).map_err(status)
        })
    }

    async fn fsinfo(&self, root_fileid: fileid3) -> Result<fsinfo3, nfsstat3> {
        let attr = match self.getattr(root_fileid).await {
            Ok(a) => post_op_attr::attributes(a),
            Err(_) => post_op_attr::Void,
        };
        Ok(fsinfo3 {
            obj_attributes: attr,
            rtmax: 1 << 20,
            rtpref: 1 << 20,
            rtmult: 4096,
            wtmax: 1 << 20,
            wtpref: 1 << 20,
            wtmult: 4096,
            dtpref: 64 << 10,
            maxfilesize: 1 << 50,
            time_delta: nfstime3 { seconds: 0, nseconds: 1 },
            properties: FSF_SYMLINK | FSF_HOMOGENEOUS | FSF_CANSETTIME,
        })
    }
}

/// Passes each connection to `front` on to the NFS server on `server`.
async fn relay(front: TcpListener, server: u16, uid: u32) {
    while let Ok((client, _)) = front.accept().await {
        tokio::spawn(async move {
            let _ = relay_one(client, server, uid).await;
        });
    }
}

/// One connection: calls made as `uid` or root (and pings) go on to the
/// server, the rest are refused; ACCESS replies say a file with no execute
/// bit can't run (macOS trusts the reply even to run a program).
async fn relay_one(client: TcpStream, server: u16, uid: u32) -> std::io::Result<()> {
    let backend = TcpStream::connect(("127.0.0.1", server)).await?;
    client.set_nodelay(true)?;
    backend.set_nodelay(true)?;
    let (mut from_client, to_client) = client.into_split();
    let (mut from_server, mut to_server) = backend.into_split();
    let to_client = Arc::new(tokio::sync::Mutex::new(to_client));
    let access: Arc<Mutex<HashSet<u32>>> = Arc::default();
    let replies = {
        let (to_client, access) = (to_client.clone(), access.clone());
        tokio::spawn(async move {
            while let Ok(Some(mut reply)) = read_message(&mut from_server).await {
                if word(&reply, 0).is_some_and(|xid| access.lock().unwrap().remove(&xid)) {
                    limit_access(&mut reply);
                }
                if write_message(&mut *to_client.lock().await, &reply).await.is_err() {
                    break;
                }
            }
        })
    };
    let result = async {
        while let Some(call) = read_message(&mut from_client).await? {
            if !allowed(&call, uid) {
                write_message(&mut *to_client.lock().await, &refusal(&call)).await?;
                continue;
            }
            if is_access(&call)
                && let Some(xid) = word(&call, 0)
            {
                access.lock().unwrap().insert(xid);
            }
            write_message(&mut to_server, &call).await?;
        }
        Ok(())
    }
    .await;
    replies.abort();
    result
}

/// The big-endian word at `at`.
fn word(message: &[u8], at: usize) -> Option<u32> {
    message.get(at..at + 4).map(|b| u32::from_be_bytes([b[0], b[1], b[2], b[3]]))
}

/// One RPC message: its fragments' bytes, joined (None at the end of the stream).
async fn read_message(stream: &mut tokio::net::tcp::OwnedReadHalf) -> std::io::Result<Option<Vec<u8>>> {
    let mut message = Vec::new();
    loop {
        let mut mark = [0u8; 4];
        match stream.read_exact(&mut mark).await {
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof && message.is_empty() => return Ok(None),
            r => r?,
        };
        let mark = u32::from_be_bytes(mark);
        let len = (mark & 0x7fff_ffff) as usize;
        if message.len() + len > MAX_MESSAGE {
            return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "RPC message too large"));
        }
        let start = message.len();
        message.resize(start + len, 0);
        stream.read_exact(&mut message[start..]).await?;
        if mark & 0x8000_0000 != 0 {
            return Ok(Some(message));
        }
    }
}

/// Sends `message` as one (last) fragment.
async fn write_message(stream: &mut tokio::net::tcp::OwnedWriteHalf, message: &[u8]) -> std::io::Result<()> {
    let framed = [&(0x8000_0000 | message.len() as u32).to_be_bytes()[..], message].concat();
    stream.write_all(&framed).await
}

/// Whether a call goes on to the server: a ping (procedure 0) always,
/// anything else only with Unix credentials naming `uid` or root.
fn allowed(call: &[u8], uid: u32) -> bool {
    // xid, CALL, RPC version, program, version, procedure, then the
    // credentials: flavor, length, and for AUTH_UNIX a stamp, the machine's
    // name (length, then padded to 4) and the uid.
    if word(call, 4) != Some(0) {
        return false;
    }
    if word(call, 20) == Some(0) {
        return true;
    }
    if word(call, 24) != Some(1) {
        return false;
    }
    let Some(name) = word(call, 36) else { return false };
    word(call, 40 + (name as usize).next_multiple_of(4)).is_some_and(|u| u == uid || u == 0)
}

/// Whether `call` is an NFS ACCESS call (program 100003, procedure 4).
fn is_access(call: &[u8]) -> bool {
    word(call, 12) == Some(100003) && word(call, 20) == Some(4)
}

/// Takes execute out of an ACCESS reply for a file with no execute bit.
fn limit_access(reply: &mut [u8]) {
    const EXECUTE: u32 = 0x20;
    // xid, REPLY, MSG_ACCEPTED, the verifier (flavor, length, body), SUCCESS,
    // NFS3_OK, attributes follow, the attributes (84 bytes: type, mode, ...),
    // then the access granted.
    let Some(verifier) = word(reply, 16) else { return };
    let at = 20 + (verifier as usize).next_multiple_of(4);
    if [word(reply, 4), word(reply, 8), word(reply, at), word(reply, at + 4), word(reply, at + 8)] != [Some(1), Some(0), Some(0), Some(0), Some(1)] {
        return;
    }
    let granted = at + 12 + 84;
    if let (Some(1), Some(mode), Some(access)) = (word(reply, at + 12), word(reply, at + 16), word(reply, granted))
        && mode & 0o111 == 0
    {
        reply[granted..granted + 4].copy_from_slice(&(access & !EXECUTE).to_be_bytes());
    }
}

/// The reply refusing `call`: denied, its credentials too weak.
fn refusal(call: &[u8]) -> Vec<u8> {
    // xid, REPLY, MSG_DENIED, AUTH_ERROR, AUTH_TOOWEAK.
    [call.get(..4).unwrap_or(&[0; 4]), &[1u32, 1, 1, 5].map(u32::to_be_bytes).concat()].concat()
}

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

    fn call(prog: u32, proc_: u32, flavor: u32, uid: u32) -> Vec<u8> {
        let mut c: Vec<u8> = [7u32, 0, 2, prog, 3, proc_, flavor].iter().flat_map(|w| w.to_be_bytes()).collect();
        // AUTH_UNIX: length, stamp, "mac" (padded), uid, gid, no groups.
        let body: Vec<u8> = [&1u32.to_be_bytes()[..], &3u32.to_be_bytes(), b"mac\0", &uid.to_be_bytes(), &20u32.to_be_bytes(), &0u32.to_be_bytes()].concat();
        c.extend((body.len() as u32).to_be_bytes());
        c.extend(body);
        c
    }

    #[test]
    fn only_the_owner_gets_in() {
        assert!(allowed(&call(100003, 1, 1, 501), 501));
        assert!(allowed(&call(100003, 1, 1, 0), 501));
        assert!(!allowed(&call(100003, 1, 1, 502), 501));
        assert!(!allowed(&call(100003, 1, 0, 501), 501));
        assert!(allowed(&call(100003, 0, 0, 999), 501));
        assert_eq!(refusal(&call(100003, 1, 1, 502))[..4], 7u32.to_be_bytes());
    }

    #[test]
    fn plain_files_dont_run() {
        let reply = |mode: u32| -> Vec<u8> {
            let mut r: Vec<u8> = [9u32, 1, 0, 0, 0, 0, 0, 1, 1, mode].iter().flat_map(|w| w.to_be_bytes()).collect();
            r.extend([0u8; 76]);
            r.extend(0x3fu32.to_be_bytes());
            r
        };
        let mut plain = reply(0o644);
        limit_access(&mut plain);
        assert_eq!(word(&plain, plain.len() - 4), Some(0x1f));
        let mut exec = reply(0o755);
        limit_access(&mut exec);
        assert_eq!(word(&exec, exec.len() - 4), Some(0x3f));
    }

    #[test]
    fn macos_files_are_bookkeeping() {
        assert!(bookkeeping("._a.txt") && bookkeeping(".DS_Store") && bookkeeping(".nfs.0001.0002"));
        assert!(!bookkeeping("a.txt") && !bookkeeping(".nfsrc") && !bookkeeping(".gitignore"));
    }
}