//! NFSv3 share consumer — in-process userspace client.
//!
//! v0.4.67 brings NFS back, this time *the right way*: a pure-Rust
//! NFSv3 client (`nfs3_client` from the xetdata family of crates)
//! that speaks the NFS protocol entirely in userspace over TCP. No
//! `mount.nfs`, no kernel modules, no `CAP_SYS_ADMIN`. Works in every
//! container the SMB path (v0.4.65) works in, including Unraid and
//! restricted-SCC OpenShift.
//!
//! ## How it differs from the v0.4.64 NFS path
//!
//! v0.4.64 shelled out to `mount(8)` and asked the kernel to attach
//! the remote share to the local filesystem. That required nfs client
//! modules on the **host** kernel. Containers that ran on hosts
//! without those modules — Unraid being the dominant case — failed
//! with `mount.nfs: failed to apply fstab options` no matter what
//! capabilities they were granted.
//!
//! v0.4.67 doesn't touch the kernel. The client opens a TCP socket to
//! the NFS server, performs the MOUNT3 RPC to get a root file handle,
//! and uses NFSv3 READDIR3 / LOOKUP3 / READ3 / GETATTR3 to walk the
//! share and stream files. The kernel sees plain TCP traffic, nothing
//! else.
//!
//! ## How it differs from the v0.4.65 SMB path
//!
//! SMB ships in v0.4.65 via the `smbclient` CLI subprocess. NFS ships
//! here via a Rust crate — no subprocess, no PATH lookup, no parsing
//! of human-formatted output. That also means:
//!
//!   - **Range requests work** for NFS-sourced ISOs. NFSv3 READ3
//!     takes an explicit offset, so the HTTP handler can seek into
//!     the middle of a 5 GB ISO without downloading what comes
//!     before. SMB-sourced ISOs still return 416 for ranges because
//!     `smbclient -c 'get file -'` is a sequential stream.
//!   - **Introspection could work** (we could LOOKUP the ISO9660 PVD
//!     at offset 0x8000 and parse the volume label). For v0.4.67 we
//!     don't yet — same as SMB, NFS-sourced ISOs register with an
//!     `Unknown` family and fall back to generic sanboot. A follow-up
//!     can add bounded read-based detection.
//!
//! ## How it's the same as SMB
//!
//! The public surface is deliberately parallel: `NfsShareManager`
//! mirrors `SmbShareManager`, `NfsShare` mirrors `SmbShare`, the API
//! responses use the same `{error, stderr, hint}` shape so the UI's
//! storage tab renders both protocols through one code path. The
//! state file (`<work_dir>/nfs_shares.json`) sits next to
//! `smb_shares.json`.
//!
//! ## No auth
//!
//! NFSv3 uses AUTH_SYS by default — the client tells the server "I'm
//! uid X, gid Y" and the server decides whether to trust that. Most
//! NAS appliances exposing ISO libraries via NFS gate access by
//! **client IP address** rather than uid, so there's nothing for the
//! UI to ask for. (If a future server needs Kerberos or non-default
//! uid mapping we can add those, but for ISO read access nobody does.)

use crate::introspect::{DistroFamily, IntrospectionReport};
use crate::store::{generate_boot_entries_for, slugify_str, IsoSource, IsoStore};
use bytes::Bytes;
use nfs3_client::tokio::TokioConnector;
use nfs3_client::Nfs3ConnectionBuilder;
use nfs3_types::nfs3::{
    self as nfs3, diropargs3, entry3, filename3, nfs_fh3, GETATTR3args, LOOKUP3args,
    Nfs3Result, READ3args, READDIR3args,
};
use nfs3_types::xdr_codec::Opaque;
use openpxe_core::{Error, Result};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::borrow::Cow;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use time::OffsetDateTime;

/// Default TCP port for the NFSv3 protocol. The classic split (111
/// portmapper + dynamic mountd) is supported by the underlying crate;
/// for direct connection-mode NAS appliances 2049 is what answers.
const DEFAULT_NFS_PORT: u16 = 2049;

/// How long we wait for the initial MOUNT3 + portmap handshake before
/// giving up. Mirrors the SMB pre-flight probe shape so the UI's
/// timeout banner reads consistently across protocols.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);

/// NFSv3 READ3 chunk size. The protocol's hard cap is 1 MiB per
/// reply; 64 KiB is a polite default that nearly every server hands
/// back without fragmentation and keeps in-flight memory bounded
/// during a streaming response.
const READ_CHUNK_BYTES: u32 = 64 * 1024;

/// Bound on the in-flight queue between the read-loop task and the
/// HTTP body stream. 16 * 64 KiB ≈ 1 MiB max buffer per stream —
/// enough to keep the network pipe full without letting a slow
/// client park gigabytes of decoded ISO in RAM.
const STREAM_BUFFER_DEPTH: usize = 16;

/// One configured NFS share. The id is derived from server+export so
/// re-adding the same coordinates is idempotent.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NfsShare {
    pub id: String,
    pub server: String,
    /// Export path on the server (e.g. "/srv/isos"). Must start with
    /// "/" to match the NFS server's view; we validate on add.
    pub export: String,
    /// TCP port — 2049 unless the operator overrode it.
    #[serde(default = "default_port")]
    pub port: u16,
    /// Most recent error talking to the share, or `None` on success.
    pub last_error: Option<String>,
    /// Operator-friendly translation of `last_error`. None when we
    /// don't have a friendlier rendition.
    pub last_hint: Option<String>,
    #[serde(with = "time::serde::rfc3339::option")]
    pub last_scan: Option<OffsetDateTime>,
    pub iso_count: u32,
    /// True after a successful scan, false on failure. Drives the
    /// UI dot.
    pub reachable: bool,
}

/// Submission from the UI / API.
#[derive(Debug, Clone, Deserialize)]
pub struct NfsAddRequest {
    pub server: String,
    pub export: String,
    #[serde(default)]
    pub port: Option<u16>,
}

fn default_port() -> u16 {
    DEFAULT_NFS_PORT
}

/// Structured error surfaced to the API and rendered in the UI. Same
/// shape as `SmbShareError` so the storage tab uses one rendering
/// path for both protocols.
#[derive(Debug, Clone, Serialize)]
pub struct NfsShareError {
    pub error: String,
    pub stderr: String,
    pub hint: Option<String>,
}

impl NfsShareError {
    fn from_raw(error: impl Into<String>, stderr: impl Into<String>) -> Self {
        let stderr = stderr.into();
        let error = error.into();
        let hint = hint_for(&stderr).or_else(|| hint_for(&error));
        Self {
            error,
            stderr,
            hint,
        }
    }
}

#[derive(Debug, Default)]
struct Inner {
    shares: HashMap<String, NfsShare>,
}

/// Manages NFS shares. Cheap to clone — internal state is
/// `Arc<Mutex<...>>`.
#[derive(Debug, Clone)]
pub struct NfsShareManager {
    state_path: Arc<PathBuf>,
    inner: Arc<Mutex<Inner>>,
    iso_store: IsoStore,
    /// Serializes scan operations on the same manager. Each scan
    /// opens its own NFS connection so concurrency isn't a hard
    /// requirement, but serializing keeps log output predictable.
    op_lock: Arc<tokio::sync::Mutex<()>>,
}

impl NfsShareManager {
    /// Construct a manager. State persists to
    /// `<work_dir>/nfs_shares.json`.
    #[must_use]
    pub fn new(work_dir: &Path, iso_store: IsoStore) -> Self {
        let state_path = work_dir.join("nfs_shares.json");
        Self {
            state_path: Arc::new(state_path),
            inner: Arc::new(Mutex::new(Inner::default())),
            iso_store,
            op_lock: Arc::new(tokio::sync::Mutex::new(())),
        }
    }

    /// Load persisted state and re-scan every share. Per-share
    /// failures are logged but never propagated — startup must not
    /// block on a single offline server.
    pub async fn load_and_rescan(&self) -> Result<()> {
        let shares = match tokio::fs::read_to_string(self.state_path.as_path()).await {
            Ok(text) => serde_json::from_str::<Vec<NfsShare>>(&text).unwrap_or_default(),
            Err(_) => Vec::new(),
        };
        for mut s in shares {
            s.last_error = None;
            s.last_hint = None;
            s.reachable = false;
            self.inner.lock().shares.insert(s.id.clone(), s.clone());
            if let Err(e) = self.rescan_inner(&s.id).await {
                tracing::warn!(
                    target: "openpxe::nfs",
                    id = %s.id, server = %s.server, export = %s.export,
                    "rescan on startup failed: {e}"
                );
            }
        }
        Ok(())
    }

    /// Register an NFS share. Validates, probes connectivity by
    /// performing a real MOUNT3 + READDIR3, and registers the
    /// resulting ISOs with the store.
    pub async fn add(
        &self,
        req: NfsAddRequest,
    ) -> std::result::Result<NfsShare, NfsShareError> {
        let server = normalize_server(&req.server);
        let export = req.export.trim().to_string();
        if server.is_empty() {
            return Err(NfsShareError::from_raw("server is required", ""));
        }
        if !export.starts_with('/') {
            return Err(NfsShareError::from_raw(
                "export path must start with '/' (e.g. /srv/isos)",
                "",
            ));
        }
        if server.contains('\0') || export.contains('\0') {
            return Err(NfsShareError::from_raw("NUL bytes are not allowed", ""));
        }
        let port = req.port.filter(|p| *p != 0).unwrap_or(DEFAULT_NFS_PORT);

        let id = share_id(&server, &export);
        let spec = NfsShare {
            id: id.clone(),
            server,
            export,
            port,
            last_error: None,
            last_hint: None,
            last_scan: None,
            iso_count: 0,
            reachable: false,
        };
        self.inner.lock().shares.insert(id.clone(), spec);
        self.persist_locked();

        if let Err(e) = self.rescan_inner(&id).await {
            let m = self.get(&id);
            return Err(NfsShareError {
                error: m.as_ref().and_then(|m| m.last_error.clone())
                    .unwrap_or_else(|| e.to_string()),
                stderr: String::new(),
                hint: m.and_then(|m| m.last_hint),
            });
        }
        Ok(self.get(&id).expect("just inserted"))
    }

    /// Remove a share. Drops every ISO sourced from it. Idempotent.
    /// Async for symmetry with [`SmbShareManager::remove`] — the
    /// SMB version is async because it tears down a credentials
    /// file; NFS has nothing to clean up but we keep the signature
    /// uniform so the call sites in app.rs / terminal.rs match.
    #[allow(clippy::unused_async)]
    pub async fn remove(&self, id: &str) -> Result<()> {
        let removed = self.inner.lock().shares.remove(id).is_some();
        if removed {
            self.iso_store.drop_external_source(id);
            self.persist_locked();
        }
        Ok(())
    }

    /// Re-walk a share for new / removed ISOs.
    pub async fn rescan(&self, id: &str) -> Result<u32> {
        self.rescan_inner(id).await
    }

    #[must_use]
    pub fn list(&self) -> Vec<NfsShare> {
        let g = self.inner.lock();
        let mut v: Vec<_> = g.shares.values().cloned().collect();
        v.sort_by(|a, b| a.id.cmp(&b.id));
        v
    }

    #[must_use]
    pub fn get(&self, id: &str) -> Option<NfsShare> {
        self.inner.lock().shares.get(id).cloned()
    }

    /// Open a byte stream reading `filename` out of share `share_id`,
    /// starting at `start_offset` and reading at most `max_len`
    /// bytes. The returned stream yields `Bytes` chunks (≤
    /// `READ_CHUNK_BYTES`) and ends at EOF, after `max_len` bytes, or
    /// on the first transport error.
    ///
    /// Used by the HTTP ISO download handler. Supports HTTP Range
    /// requests because NFSv3 READ3 takes an explicit offset — this
    /// is the protocol-level advantage NFS has over the SMB
    /// userspace path.
    ///
    /// Kept `async` for symmetry with [`SmbShareManager::stream_iso`]
    /// even though the body doesn't await today — a future
    /// refinement (e.g. throttling, connection pooling) will need to
    /// await without changing call sites.
    #[allow(clippy::unused_async)]
    pub async fn stream_iso(
        &self,
        share_id: &str,
        filename: &str,
        start_offset: u64,
        max_len: Option<u64>,
    ) -> Result<NfsStream> {
        let share = self
            .get(share_id)
            .ok_or_else(|| Error::Invalid(format!("no such NFS share '{share_id}'")))?;
        // Defensive: NFSv3 LOOKUP3 takes a single name relative to
        // the export root, not a path. We don't support nested
        // directories in v0.4.67 — ISOs live at the top of the share.
        if filename.contains('/') || filename.contains('\\') || filename.contains("..") {
            return Err(Error::Invalid(format!("invalid filename '{filename}'")));
        }

        let (tx, rx) =
            tokio::sync::mpsc::channel::<std::io::Result<Bytes>>(STREAM_BUFFER_DEPTH);
        let server = share.server.clone();
        let export = share.export.clone();
        let port = share.port;
        let fname = filename.to_string();

        // Spawn a task that owns the NFS connection. Owning the
        // connection inside the spawn means we don't have to worry
        // about borrowing across awaits or sharing the connection
        // between scan and stream — each stream gets its own.
        let task = tokio::spawn(async move {
            let result = stream_loop(
                &server,
                &export,
                port,
                &fname,
                start_offset,
                max_len,
                tx.clone(),
            )
            .await;
            if let Err(e) = result {
                // Best-effort signal of the error to the consumer.
                // If the receiver has already dropped we just exit.
                let _ = tx
                    .send(Err(std::io::Error::other(e.to_string())))
                    .await;
            }
        });

        Ok(NfsStream {
            rx,
            _task: task,
        })
    }

    // ── internals ─────────────────────────────────────────────────────

    async fn rescan_inner(&self, id: &str) -> Result<u32> {
        let _g = self.op_lock.lock().await;
        let share = self
            .get(id)
            .ok_or_else(|| Error::Invalid(format!("no such share '{id}'")))?;
        let now = OffsetDateTime::now_utc();

        // Drop prior entries so a deleted file disappears from the
        // store on the next scan.
        self.iso_store.drop_external_source(id);

        let listing = match list_isos(&share.server, &share.export, share.port).await {
            Ok(l) => l,
            Err(err) => {
                let stderr = err.to_string();
                let hint = hint_for(&stderr);
                self.update_status(id, 0, false, Some(stderr.clone()), hint, now);
                return Err(Error::Invalid(stderr));
            }
        };

        let mut count = 0u32;
        for entry in listing {
            let iso_id = format!("nfs-{}-{}", share.id, slugify_str(&entry.filename));
            // Same approach as SMB: no real introspection over the
            // network in v0.4.67. The boot-entry generator falls back
            // to filename-based sanboot detection.
            let report = IntrospectionReport {
                family: DistroFamily::Unknown,
                volume_label: None,
                kernel_path: None,
                initrd_paths: Vec::new(),
                has_boot_wim: false,
            };
            let boot_entries = generate_boot_entries_for(&iso_id, &entry.filename, &report);
            let source = IsoSource::Nfs {
                share_id: share.id.clone(),
                relative_path: entry.filename.clone(),
            };
            self.iso_store.register_external(
                iso_id,
                entry.filename,
                entry.size,
                report,
                boot_entries,
                source,
            );
            count += 1;
        }
        self.update_status(id, count, true, None, None, now);
        tracing::info!(
            target: "openpxe::nfs",
            id = %id, server = %share.server, export = %share.export,
            iso_count = count,
            "NFS share scanned"
        );
        Ok(count)
    }

    fn update_status(
        &self,
        id: &str,
        iso_count: u32,
        reachable: bool,
        err: Option<String>,
        hint: Option<String>,
        ts: OffsetDateTime,
    ) {
        if let Some(s) = self.inner.lock().shares.get_mut(id) {
            s.iso_count = iso_count;
            s.reachable = reachable;
            s.last_error = err;
            s.last_hint = hint;
            s.last_scan = Some(ts);
        }
        self.persist_locked();
    }

    fn persist_locked(&self) {
        let shares: Vec<NfsShare> = self.inner.lock().shares.values().cloned().collect();
        let path = self.state_path.as_path();
        let tmp = path.with_extension("json.tmp");
        let body = match serde_json::to_vec_pretty(&shares) {
            Ok(b) => b,
            Err(e) => {
                tracing::warn!(target: "openpxe::nfs", "serialize: {e}");
                return;
            }
        };
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        if let Err(e) = std::fs::write(&tmp, body) {
            tracing::warn!(target: "openpxe::nfs", "write tmp: {e}");
            return;
        }
        if let Err(e) = std::fs::rename(&tmp, path) {
            tracing::warn!(target: "openpxe::nfs", "rename: {e}");
        }
    }
}

/// HTTP body stream for an NFS-sourced ISO read.
///
/// Implements `Stream<Item = io::Result<Bytes>>` so axum can convert
/// it into a response body via `Body::from_stream`.
#[derive(Debug)]
pub struct NfsStream {
    rx: tokio::sync::mpsc::Receiver<std::io::Result<Bytes>>,
    /// Kept alive so the read-loop task isn't dropped while the HTTP
    /// client is still consuming bytes. Dropping the stream cancels
    /// the task, which is the right behavior on client disconnect.
    _task: tokio::task::JoinHandle<()>,
}

impl futures::Stream for NfsStream {
    type Item = std::io::Result<Bytes>;
    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        self.rx.poll_recv(cx)
    }
}

#[derive(Debug, Clone)]
struct NfsListEntry {
    filename: String,
    size: u64,
}

/// Connect, READDIR the export root, look up each `*.iso` to get its
/// size + file handle. Returns a flat list. Errors are returned with
/// a human-readable message; the caller decides how to surface them.
async fn list_isos(
    server: &str,
    export: &str,
    port: u16,
) -> std::result::Result<Vec<NfsListEntry>, NfsClientError> {
    // build_connection applies its own per-attempt timeout (privileged
    // source port first, then a non-privileged fallback).
    let mut conn = build_connection(server, export, port).await?;

    let root = conn.root_nfs_fh3();
    let mut entries = Vec::new();
    let mut cookie: u64 = 0;
    let mut cookieverf = nfs3::cookieverf3::default();
    loop {
        let res = conn
            .readdir(&READDIR3args {
                dir: root.clone(),
                cookie,
                cookieverf,
                count: 32 * 1024,
            })
            .await
            .map_err(NfsClientError::Rpc)?;
        let ok = match res {
            Nfs3Result::Ok(ok) => ok,
            Nfs3Result::Err((status, _)) => {
                return Err(NfsClientError::Nfsstat(status_label(status)));
            }
        };
        cookieverf = ok.cookieverf;
        let eof = ok.reply.eof;
        let dir_entries: Vec<entry3<'_>> = ok.reply.entries.0;
        let next_cookie = dir_entries.last().map(|e| e.cookie);
        for entry in dir_entries {
            // entry.name is `filename3<'a>(Opaque<'a>)` — XDR opaque
            // bytes. Decode to UTF-8 best-effort.
            let name_bytes = entry.name.0.as_ref();
            let Ok(name) = std::str::from_utf8(name_bytes) else {
                continue;
            };
            if name == "." || name == ".." {
                continue;
            }
            if !name.to_ascii_lowercase().ends_with(".iso") {
                continue;
            }
            // Look up the file to get its size + verify it's a
            // regular file (not a symlink / directory matching the
            // .iso pattern).
            let lookup = conn
                .lookup(&LOOKUP3args {
                    what: diropargs3 {
                        dir: root.clone(),
                        name: filename3(Opaque::borrowed(name_bytes)),
                    },
                })
                .await
                .map_err(NfsClientError::Rpc)?;
            let lookup_ok = match lookup {
                Nfs3Result::Ok(o) => o,
                Nfs3Result::Err(_) => continue,
            };
            let getattr = conn
                .getattr(&GETATTR3args {
                    object: lookup_ok.object,
                })
                .await
                .map_err(NfsClientError::Rpc)?;
            let attrs = match getattr {
                Nfs3Result::Ok(o) => o.obj_attributes,
                Nfs3Result::Err(_) => continue,
            };
            // ftype3::NF3REG == 1 (regular file). Skip everything
            // else — directories, symlinks, devices.
            if attrs.type_ as u32 != nfs3::ftype3::NF3REG as u32 {
                continue;
            }
            entries.push(NfsListEntry {
                filename: name.to_string(),
                size: attrs.size,
            });
        }
        if eof {
            break;
        }
        match next_cookie {
            Some(c) if c != 0 => cookie = c,
            _ => break,
        }
    }

    let _ = conn.unmount().await;
    Ok(entries)
}

async fn stream_loop(
    server: &str,
    export: &str,
    port: u16,
    filename: &str,
    start_offset: u64,
    max_len: Option<u64>,
    tx: tokio::sync::mpsc::Sender<std::io::Result<Bytes>>,
) -> std::result::Result<(), NfsClientError> {
    let mut conn = build_connection(server, export, port).await?;

    let root = conn.root_nfs_fh3();
    // Look up the file to get its handle.
    let name_bytes = filename.as_bytes();
    let lookup = conn
        .lookup(&LOOKUP3args {
            what: diropargs3 {
                dir: root,
                name: filename3(Opaque::borrowed(name_bytes)),
            },
        })
        .await
        .map_err(NfsClientError::Rpc)?;
    let lookup_ok = match lookup {
        Nfs3Result::Ok(o) => o,
        Nfs3Result::Err((status, _)) => {
            return Err(NfsClientError::Nfsstat(status_label(status)));
        }
    };
    let file_handle: nfs_fh3 = lookup_ok.object;

    let mut offset = start_offset;
    let mut remaining = max_len;
    loop {
        if let Some(r) = remaining {
            if r == 0 {
                break;
            }
        }
        // Cap the chunk at READ_CHUNK_BYTES and at the remaining budget.
        let chunk = match remaining {
            Some(r) if r < u64::from(READ_CHUNK_BYTES) => r as u32,
            _ => READ_CHUNK_BYTES,
        };
        let res = conn
            .read(&READ3args {
                file: file_handle.clone(),
                offset,
                count: chunk,
            })
            .await
            .map_err(NfsClientError::Rpc)?;
        let ok = match res {
            Nfs3Result::Ok(o) => o,
            Nfs3Result::Err((status, _)) => {
                return Err(NfsClientError::Nfsstat(status_label(status)));
            }
        };
        let bytes = Bytes::copy_from_slice(ok.data.as_ref());
        let bytes_len = bytes.len() as u64;
        if tx.send(Ok(bytes)).await.is_err() {
            // HTTP client dropped — abort gracefully.
            break;
        }
        offset += bytes_len;
        if let Some(r) = remaining.as_mut() {
            *r = r.saturating_sub(bytes_len);
        }
        if ok.eof {
            break;
        }
        // Defensive: a server that returns 0 bytes without EOF
        // would have us busy-looping. Bail out instead.
        if bytes_len == 0 {
            break;
        }
    }

    let _ = conn.unmount().await;
    Ok(())
}

/// Mount the export and return a live connection.
///
/// ## Privileged source port (the v0.4.68 fix)
///
/// Linux kernel `nfsd` — which is what UniFi UNAS, Synology, TrueNAS,
/// and essentially every appliance NAS runs underneath — exports with
/// the `secure` option **by default**. `secure` means the server only
/// accepts MOUNT3 / NFS3 requests whose TCP **source** port is in the
/// privileged range (< 1024). A client connecting from an ephemeral
/// high port gets `MNT3ERR_ACCES` at mount time — which is exactly the
/// error operators hit in v0.4.67 even with their host IP correctly in
/// the export's allow-list.
///
/// v0.4.67 disabled privileged source ports because the openpxe
/// process runs as uid 10001 and "can't bind sub-1024 ports". That
/// reasoning was wrong: the binary carries `CAP_NET_BIND_SERVICE`
/// (granted via `setcap` in the Dockerfile so it can bind the DHCP /
/// TFTP / HTTP low ports as non-root), and that capability also lets
/// it bind a privileged *source* port for an outbound connection.
///
/// So we now try a privileged source port first — the common case for
/// real NAS appliances — and fall back to a non-privileged port for
/// servers exported `insecure` (or environments where we genuinely
/// can't grab a low port). Each attempt gets its own connect timeout.
async fn build_connection(
    server: &str,
    export: &str,
    port: u16,
) -> std::result::Result<
    nfs3_client::Nfs3Connection<nfs3_client::tokio::TokioIo<tokio::net::TcpStream>>,
    NfsClientError,
> {
    match connect_once(server, export, port, true).await {
        Ok(conn) => Ok(conn),
        // A timeout means the server didn't answer at all — a
        // non-privileged retry would just time out again and double
        // the operator's wait. Surface the timeout immediately.
        Err(primary @ NfsClientError::Timeout(..)) => Err(primary),
        Err(primary) => match connect_once(server, export, port, false).await {
            Ok(conn) => Ok(conn),
            // Surface the privileged-attempt error: for the dominant
            // `secure`-export case it's the one whose hint points at
            // the real fix.
            Err(_) => Err(primary),
        },
    }
}

/// One mount attempt with a specific source-port policy, bounded by
/// [`CONNECT_TIMEOUT`].
async fn connect_once(
    server: &str,
    export: &str,
    port: u16,
    privileged: bool,
) -> std::result::Result<
    nfs3_client::Nfs3Connection<nfs3_client::tokio::TokioIo<tokio::net::TcpStream>>,
    NfsClientError,
> {
    let fut = Nfs3ConnectionBuilder::new(TokioConnector, server, export)
        .connect_from_privileged_port(privileged)
        .nfs3_port(port)
        .mount();
    match tokio::time::timeout(CONNECT_TIMEOUT, fut).await {
        Ok(Ok(conn)) => Ok(conn),
        Ok(Err(e)) => Err(NfsClientError::Connect(e.to_string())),
        Err(_) => Err(NfsClientError::Timeout(server.to_string(), port)),
    }
}

#[derive(Debug)]
enum NfsClientError {
    Connect(String),
    Timeout(String, u16),
    Rpc(nfs3_client::RpcError),
    Nfsstat(String),
}

impl std::fmt::Display for NfsClientError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Connect(msg) => write!(f, "connect failed: {msg}"),
            Self::Timeout(host, port) => write!(
                f,
                "connect timed out after {}s talking to {host}:{port}",
                CONNECT_TIMEOUT.as_secs()
            ),
            Self::Rpc(e) => write!(f, "RPC failure: {e}"),
            Self::Nfsstat(s) => write!(f, "NFS server returned {s}"),
        }
    }
}

impl std::error::Error for NfsClientError {}

/// Best-effort label for an `nfsstat3` so the UI shows a readable
/// name like `NFS3ERR_ACCES` instead of a magic number.
fn status_label(status: nfs3::nfsstat3) -> String {
    use nfs3::nfsstat3 as S;
    let name = match status {
        S::NFS3_OK => "NFS3_OK",
        S::NFS3ERR_PERM => "NFS3ERR_PERM",
        S::NFS3ERR_NOENT => "NFS3ERR_NOENT",
        S::NFS3ERR_IO => "NFS3ERR_IO",
        S::NFS3ERR_NXIO => "NFS3ERR_NXIO",
        S::NFS3ERR_ACCES => "NFS3ERR_ACCES",
        S::NFS3ERR_EXIST => "NFS3ERR_EXIST",
        S::NFS3ERR_XDEV => "NFS3ERR_XDEV",
        S::NFS3ERR_NODEV => "NFS3ERR_NODEV",
        S::NFS3ERR_NOTDIR => "NFS3ERR_NOTDIR",
        S::NFS3ERR_ISDIR => "NFS3ERR_ISDIR",
        S::NFS3ERR_INVAL => "NFS3ERR_INVAL",
        S::NFS3ERR_FBIG => "NFS3ERR_FBIG",
        S::NFS3ERR_NOSPC => "NFS3ERR_NOSPC",
        S::NFS3ERR_ROFS => "NFS3ERR_ROFS",
        S::NFS3ERR_MLINK => "NFS3ERR_MLINK",
        S::NFS3ERR_NAMETOOLONG => "NFS3ERR_NAMETOOLONG",
        S::NFS3ERR_NOTEMPTY => "NFS3ERR_NOTEMPTY",
        S::NFS3ERR_DQUOT => "NFS3ERR_DQUOT",
        S::NFS3ERR_STALE => "NFS3ERR_STALE",
        S::NFS3ERR_REMOTE => "NFS3ERR_REMOTE",
        S::NFS3ERR_BADHANDLE => "NFS3ERR_BADHANDLE",
        S::NFS3ERR_NOT_SYNC => "NFS3ERR_NOT_SYNC",
        S::NFS3ERR_BAD_COOKIE => "NFS3ERR_BAD_COOKIE",
        S::NFS3ERR_NOTSUPP => "NFS3ERR_NOTSUPP",
        S::NFS3ERR_TOOSMALL => "NFS3ERR_TOOSMALL",
        S::NFS3ERR_SERVERFAULT => "NFS3ERR_SERVERFAULT",
        S::NFS3ERR_BADTYPE => "NFS3ERR_BADTYPE",
        S::NFS3ERR_JUKEBOX => "NFS3ERR_JUKEBOX",
    };
    name.to_string()
}

/// Translate well-known NFS error strings into actionable hints for
/// the UI. Mirrors the SMB hint table in spirit; the patterns are
/// different because NFS errors travel via NFS3ERR_* codes plus
/// transport-level messages from the Rust crate.
fn hint_for(text: &str) -> Option<String> {
    let s = text.to_ascii_lowercase();
    if s.contains("mnt3err_acces") || s.contains("mount") && s.contains("acces") {
        // Mount-protocol access denial. Two common causes, in order of
        // likelihood for an appliance NAS: (1) the export requires a
        // privileged source port (`secure`, the Linux default) — we
        // already retry with one, so reaching here means even that was
        // refused; (2) the client IP isn't in the allow-list.
        Some(
            "the NFS server denied the mount (MNT3ERR_ACCES). Two things to \
             check on the server: (1) this OpenPXE host's IP is in the \
             export's allowed-clients list, and (2) if your export uses the \
             default `secure` option, OpenPXE already connects from a \
             privileged port — but if the server still refuses, add \
             `insecure` to the export. On UniFi UNAS, confirm the host IP is \
             listed under the share's NFS permissions and the export path is \
             /var/nfs/shared/<share> (not just /<share>)."
                .into(),
        )
    } else if s.contains("nfs3err_acces") || s.contains("permission denied") {
        Some(
            "the NFS server rejected this client. Most likely your export \
             is restricted by client IP — add this OpenPXE host (or its \
             subnet) to the export's allowed-clients list on the server."
                .into(),
        )
    } else if s.contains("nfs3err_noent")
        || s.contains("nfs3err_notdir")
        || s.contains("mnt3err_noent")
    {
        Some(
            "the export path doesn't exist on the server, or it isn't a \
             directory. Double-check the path (e.g. /srv/isos vs /isos — \
             UniFi UNAS Pro exposes shares under /var/nfs/shared/<name>)."
                .into(),
        )
    } else if s.contains("nfs3err_stale") || s.contains("nfs3err_badhandle") {
        Some(
            "the server's view of the share changed under us. Re-scan; \
             if that doesn't help, remove and re-add the share."
                .into(),
        )
    } else if s.contains("connect timed out")
        || s.contains("timed out")
        || s.contains("connection timed out")
    {
        Some(
            "no answer from the server within the connect timeout. \
             Verify the IP, the port (default 2049), and any firewall \
             between OpenPXE and the NAS."
                .into(),
        )
    } else if s.contains("connection refused") {
        Some(
            "the NFS service isn't accepting connections on this port. \
             Make sure nfsd is running on the server and (for v3) the \
             portmapper on port 111 is reachable."
                .into(),
        )
    } else if s.contains("no route to host") || s.contains("network is unreachable") {
        Some(
            "the server isn't reachable on this network. Check the IP \
             and routes."
                .into(),
        )
    } else if s.contains("mount") && s.contains("denied") {
        Some(
            "MOUNT3 was denied. The classic cause is that the export's \
             `rw=<host>` / `ro=<host>` list doesn't include this client. \
             Some servers also require `insecure` in /etc/exports for \
             non-privileged source ports — which is what this client \
             uses (we run as uid 10001, no privileged-port capability)."
                .into(),
        )
    } else {
        None
    }
}

fn share_id(server: &str, export: &str) -> String {
    slugify_str(&format!("{server}{export}"))
}

/// Normalize a server input: trim, strip schemes, drop trailing
/// slashes. Matches the SMB normalizer so paste-from-anywhere works.
fn normalize_server(raw: &str) -> String {
    let s = raw.trim();
    let s = s
        .strip_prefix("nfs://")
        .or_else(|| s.strip_prefix("http://"))
        .or_else(|| s.strip_prefix("https://"))
        .unwrap_or(s);
    s.trim_end_matches('/').to_string()
}

// Silence the unused-import warning on `Cow` — we use it implicitly
// via `Opaque::borrowed` constructions in `list_isos`.
#[allow(dead_code)]
fn _unused() -> Cow<'static, str> {
    Cow::Borrowed("")
}

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

    #[test]
    fn share_id_is_stable_and_safe() {
        let a = share_id("10.0.0.5", "/srv/isos");
        let b = share_id("10.0.0.5", "/srv/isos");
        assert_eq!(a, b);
        assert!(!a.contains('/'));
        assert!(!a.contains('.'));
    }

    #[test]
    fn normalize_server_strips_schemes() {
        assert_eq!(normalize_server("  10.0.0.5  "), "10.0.0.5");
        assert_eq!(normalize_server("nfs://nas.lan/"), "nas.lan");
        assert_eq!(normalize_server("http://192.168.1.51"), "192.168.1.51");
        assert_eq!(normalize_server("nas.lan"), "nas.lan");
    }

    #[test]
    fn hint_for_acces_points_to_exports_table() {
        let h = hint_for("NFS server returned NFS3ERR_ACCES").unwrap();
        assert!(
            h.to_lowercase().contains("export"),
            "expected exports guidance, got: {h}"
        );
    }

    #[test]
    fn hint_for_mount_acces_calls_out_privileged_port_and_allowlist() {
        // The dominant v0.4.67 field failure: mount denied even with the
        // host IP allow-listed, because the export is `secure` and the
        // client used a high source port. The hint should mention both
        // the allow-list and the secure/insecure angle.
        let h = hint_for("connect failed: MNT3ERR_ACCES").unwrap();
        let lc = h.to_lowercase();
        assert!(lc.contains("insecure") || lc.contains("privileged"), "got: {h}");
        assert!(lc.contains("allow") || lc.contains("permission"), "got: {h}");
    }

    #[test]
    fn hint_for_noent_points_to_export_path() {
        let h = hint_for("NFS server returned NFS3ERR_NOENT").unwrap();
        assert!(
            h.to_lowercase().contains("path") || h.to_lowercase().contains("directory"),
            "expected export-path guidance, got: {h}"
        );
    }

    #[test]
    fn hint_for_mount_denied_calls_out_insecure_option() {
        let h = hint_for("mount denied").unwrap();
        assert!(h.to_lowercase().contains("insecure"));
    }

    #[test]
    fn hint_for_unknown_is_none() {
        assert!(hint_for("some entirely unrelated string").is_none());
    }

    #[test]
    fn status_label_covers_common_codes() {
        assert_eq!(status_label(nfs3::nfsstat3::NFS3_OK), "NFS3_OK");
        assert_eq!(status_label(nfs3::nfsstat3::NFS3ERR_ACCES), "NFS3ERR_ACCES");
        assert_eq!(status_label(nfs3::nfsstat3::NFS3ERR_NOENT), "NFS3ERR_NOENT");
    }
}