//! FTP backend.
//!
//! Repository objects are already sealed client-side, so plain FTP only
//! exposes what any storage host sees anyway; the notable residual risk
//! is that classic FTP sends the *credentials* in the clear, so anyone
//! on the wire can gain the access needed to delete backups.
//!
//! URL form: `ftp://user:password@host:port/path`. Credentials are
//! resolved most-specific first: the URL itself, `$BEEPING_FTP_PASSWORD`,
//! then a matching lftp bookmark (see [`crate::lftp`]); with nothing
//! found, the login is `anonymous` with an empty password — plenty of
//! NAS boxes serve exactly that. Writing `ftp://anonymous@host/...`
//! forces anonymous login even when a bookmark holds named credentials
//! for the host.
//!
//! At most `?connections=N` (default 4) connections are held at once,
//! and a server that refuses further logins (421) is waited out rather
//! than treated as an error — see [`crate::pool`].

use std::sync::{
    Mutex,
    atomic::{AtomicBool, Ordering},
};

use repository::{Backend, ChunkId, Error, ObjectKey, ObjectKind};
use suppaftp::{FtpError, FtpStream, Status, types::FileType};
use url::Url;

use crate::{
    lftp, percent_decode,
    pool::{Pool, temp_suffix},
};

const DEFAULT_CONNECTIONS: usize = 4;

pub struct FtpBackend {
    host: String,
    port: u16,
    user: String,
    password: String,
    root: String,
    pool: Pool<FtpStream>,
    /// Directories already created this session, to avoid a MKD round
    /// trip per stored object.
    known_dirs: Mutex<std::collections::HashSet<String>>,
    /// Set once the server has declined to list machine-readably, which
    /// is the only way to learn what the repository occupies. Asking
    /// again would cost a retry and a fresh dial for the same refusal.
    size_unknown: AtomicBool,
}

impl FtpBackend {
    pub fn from_url(url: &Url) -> Result<FtpBackend, Error> {
        let host = url
            .host_str()
            .ok_or_else(|| Error::Backend(format!("{url} has no host")))?
            .to_string();

        let port = url.port().unwrap_or(21);

        let mut user = match url.username() {
            "" => None,
            user => Some(percent_decode(user)),
        };

        let mut password = url.password().map(percent_decode);

        if password.is_none()
            && let Ok(env) = std::env::var("BEEPING_FTP_PASSWORD")
        {
            password = Some(env);
        }

        if (user.is_none() || password.is_none())
            && let Some(found) = lftp::find_credentials("ftp", &host, port, user.as_deref())
        {
            user.get_or_insert(found.user);

            if password.is_none() {
                password = found.password;
            }
        }

        let user = user.unwrap_or_else(|| "anonymous".to_string());
        let password = password.unwrap_or_default();

        Ok(FtpBackend {
            host,
            port,
            user,
            password,
            root: url.path().trim_end_matches('/').to_string(),
            pool: Pool::new(crate::connection_limit(url, DEFAULT_CONNECTIONS)?),
            size_unknown: AtomicBool::new(false),
            known_dirs: Mutex::new(std::collections::HashSet::new()),
        })
    }

    fn with_connection<T>(
        &self,
        op: impl FnMut(&mut FtpStream) -> Result<T, Error>,
    ) -> Result<T, Error> {
        self.pool.with(|| self.connect(), is_server_full, op)
    }

    fn connect(&self) -> Result<FtpStream, Error> {
        let mut ftp = FtpStream::connect((self.host.as_str(), self.port)).map_err(ftp_error)?;

        ftp.login(&self.user, &self.password).map_err(ftp_error)?;
        ftp.transfer_type(FileType::Binary).map_err(ftp_error)?;

        Ok(ftp)
    }

    fn object_path(&self, key: &ObjectKey) -> String {
        let mut path = self.root.clone();

        for segment in key.segments() {
            path.push('/');
            path.push_str(&segment);
        }

        path
    }

    fn dir_path(&self, segments: &[&str]) -> String {
        let mut path = self.root.clone();

        for segment in segments {
            path.push('/');
            path.push_str(segment);
        }

        path
    }

    /// Creates the directories leading to `key` — the repository root
    /// and every intermediate — remembering successes. MKD failures are
    /// ignored here: "already exists" is not reliably distinguishable
    /// in FTP, and a real problem surfaces as a failed STOR immediately
    /// after.
    fn ensure_parents(&self, ftp: &mut FtpStream, key: &ObjectKey) {
        let segments = key.segments();
        let mut directories = Vec::new();

        if !self.root.is_empty() {
            directories.push(self.root.clone());
        }

        let mut path = self.root.clone();

        // All but the last segment are directories
        for segment in &segments[..segments.len() - 1] {
            path.push('/');
            path.push_str(segment);
            directories.push(path.clone());
        }

        for directory in directories {
            let mut known = self.known_dirs.lock().unwrap();

            if known.contains(&directory) {
                continue;
            }

            let _ = ftp.mkdir(&directory);
            known.insert(directory);
        }
    }
}

impl Backend for FtpBackend {
    fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<(), Error> {
        let path = self.object_path(key);

        self.with_connection(|ftp| {
            if ftp.size(&path).is_ok() {
                // Objects are write-once; it is already this one
                return Ok(());
            }

            self.ensure_parents(ftp, key);

            // Upload beside the final name, then rename into place
            // so no reader (or crash) ever sees a partial object
            let temp = format!("{path}{}", temp_suffix());

            ftp.put_file(&temp, &mut std::io::Cursor::new(data))
                .map_err(ftp_error)?;

            match ftp.rename(&temp, &path) {
                Ok(()) => Ok(()),
                Err(_) if ftp.size(&path).is_ok() => {
                    // A concurrent writer beat us to it; ours is
                    // surplus
                    let _ = ftp.rm(&temp);

                    Ok(())
                }
                Err(err) => Err(ftp_error(err)),
            }
        })
    }

    fn get(&self, key: &ObjectKey) -> Result<Option<Vec<u8>>, Error> {
        let path = self.object_path(key);

        self.with_connection(|ftp| match ftp.retr_as_buffer(&path) {
            Ok(buffer) => Ok(Some(buffer.into_inner())),
            Err(err) if is_not_found(&err) => Ok(None),
            Err(err) => Err(ftp_error(err)),
        })
    }

    fn contains(&self, key: &ObjectKey) -> Result<bool, Error> {
        let path = self.object_path(key);

        self.with_connection(|ftp| match ftp.size(&path) {
            Ok(_) => Ok(true),
            Err(err) if is_not_found(&err) => Ok(false),
            Err(err) => Err(ftp_error(err)),
        })
    }

    fn list(
        &self,
        kind: ObjectKind,
        visit: &mut dyn FnMut(ObjectKey) -> Result<(), Error>,
    ) -> Result<(), Error> {
        match kind {
            ObjectKind::Header => {
                if self.contains(&ObjectKey::Header)? {
                    visit(ObjectKey::Header)?;
                }
            }
            // The flat kinds: one directory of hex-named objects each
            ObjectKind::Snapshot | ObjectKind::Prune | ObjectKind::Lock => {
                for name in self.dir_names(&self.dir_path(&[kind.directory()]))? {
                    if let Some(key) = kind.key_for(&name) {
                        visit(key)?;
                    }
                }
            }
            ObjectKind::Chunk => {
                for prefix in self.dir_names(&self.dir_path(&["chunks"]))? {
                    if prefix.len() != 2 || !prefix.bytes().all(|b| b.is_ascii_hexdigit()) {
                        continue;
                    }

                    for name in self.dir_names(&self.dir_path(&["chunks", &prefix]))? {
                        if let Ok(id) = ChunkId::from_hex(name) {
                            visit(ObjectKey::Chunk(id))?;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Adds up everything under the repository directory, from MLSD
    /// listings — one per directory, so a few hundred round trips for
    /// the fanned-out chunk store. Asked once a run, never on a hot
    /// path.
    ///
    /// MLSD is the only listing FTP has whose sizes can be read without
    /// guessing at a server's chosen `ls` format. A server without it
    /// cannot say what it holds, and says so once rather than being
    /// asked again.
    fn used_space(&self) -> Result<Option<u64>, Error> {
        if self.size_unknown.load(Ordering::Acquire) {
            return Ok(None);
        }

        let root = self.root.clone();

        match self.with_connection(|ftp| tree_bytes(ftp, &root)) {
            Ok(total) => Ok(Some(total)),
            Err(_) => {
                self.size_unknown.store(true, Ordering::Release);

                Ok(None)
            }
        }
    }

    fn delete(&self, key: &ObjectKey) -> Result<(), Error> {
        let path = self.object_path(key);

        self.with_connection(|ftp| match ftp.rm(&path) {
            Ok(()) => Ok(()),
            Err(err) if is_not_found(&err) => Ok(()),
            Err(err) => Err(ftp_error(err)),
        })
    }
}

impl FtpBackend {
    /// The entry names within a directory, treating a missing directory
    /// as empty. Servers differ on whether NLST returns bare names or
    /// full paths, so paths are reduced to their final component.
    fn dir_names(&self, path: &str) -> Result<Vec<String>, Error> {
        self.with_connection(|ftp| match ftp.nlst(Some(path)) {
            Ok(names) => Ok(names
                .into_iter()
                .map(|name| {
                    name.rsplit('/')
                        .next()
                        .expect("rsplit always yields at least one part")
                        .to_string()
                })
                .collect()),
            Err(err) if is_not_found(&err) => Ok(Vec::new()),
            Err(err) => Err(ftp_error(err)),
        })
    }
}

/// How many bytes the tree under `path` holds, by the sizes MLSD
/// reports. A directory that is not there holds nothing — the
/// repository's own subdirectories appear only once something has been
/// put in them.
fn tree_bytes(ftp: &mut FtpStream, path: &str) -> Result<u64, Error> {
    let lines = match ftp.mlsd(Some(path)) {
        Ok(lines) => lines,
        Err(err) if is_not_found(&err) => return Ok(0),
        Err(err) => return Err(ftp_error(err)),
    };

    let mut total = 0;

    for line in lines {
        let Some(listed) = Listed::parse(&line) else {
            continue;
        };

        total += match listed.kind {
            // Servers differ on whether the name is bare or a full
            // path, exactly as they do for NLST
            ListedKind::Directory => tree_bytes(ftp, &format!("{path}/{}", listed.name))?,
            ListedKind::File => listed.size,
        };
    }

    Ok(total)
}

/// One entry of an MLSD listing: `fact=value;...; name`, per RFC 3659.
struct Listed<'a> {
    kind: ListedKind,
    size: u64,
    name: &'a str,
}

enum ListedKind {
    File,
    Directory,
}

impl<'a> Listed<'a> {
    /// Reads one listing line, or `None` for the entries that are not
    /// objects of their own: the directory itself, its parent, and
    /// anything whose type the server did not give.
    fn parse(line: &'a str) -> Option<Listed<'a>> {
        let (facts, name) = line.trim_end().split_once(' ')?;

        let mut kind = None;
        let mut size = 0;

        for fact in facts.split(';').filter(|fact| !fact.is_empty()) {
            let Some((key, value)) = fact.split_once('=') else {
                continue;
            };

            match key.to_ascii_lowercase().as_str() {
                "type" => {
                    kind = match value.to_ascii_lowercase().as_str() {
                        "file" => Some(ListedKind::File),
                        "dir" => Some(ListedKind::Directory),
                        // cdir and pdir are this directory and its
                        // parent, listed for navigation
                        _ => return None,
                    };
                }
                "size" => size = value.parse().unwrap_or(0),
                _ => {}
            }
        }

        Some(Listed {
            kind: kind?,
            size,
            name: name.rsplit('/').next().expect("rsplit yields a part"),
        })
    }
}

/// FTP reports a missing file as 550; the code is also used for
/// permission problems, but that ambiguity is inherent to the protocol.
fn is_not_found(err: &FtpError) -> bool {
    matches!(
        err,
        FtpError::UnexpectedResponse(response) if response.status == Status::FileUnavailable
    )
}

/// Maps FTP errors into backend errors, giving 421 — the server
/// declining service, most commonly "too many users already logged
/// in" — its own variant so the connection pool can wait it out rather
/// than fail.
fn ftp_error(err: FtpError) -> Error {
    if matches!(
        &err,
        FtpError::UnexpectedResponse(response) if response.status == Status::NotAvailable
    ) {
        return Error::BackendBusy(format!("ftp: {err}"));
    }

    Error::Backend(format!("ftp: {err}"))
}

fn is_server_full(err: &Error) -> bool {
    matches!(err, Error::BackendBusy(_))
}

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

    // These use a fictional host so no real lftp bookmark can match,
    // and they tolerate $BEEPING_FTP_PASSWORD being set in the
    // developer's environment rather than mutating process env.

    #[test]
    fn credentialless_urls_default_to_anonymous() {
        let url = Url::parse("ftp://nas.invalid/backups/repo").unwrap();
        let backend = FtpBackend::from_url(&url).unwrap();

        assert_eq!(backend.user, "anonymous");

        if std::env::var("BEEPING_FTP_PASSWORD").is_err() {
            assert_eq!(backend.password, "");
        }
    }

    #[test]
    fn an_explicit_anonymous_user_gets_an_empty_password() {
        let url = Url::parse("ftp://anonymous@nas.invalid/backups/repo").unwrap();
        let backend = FtpBackend::from_url(&url).unwrap();

        assert_eq!(backend.user, "anonymous");

        if std::env::var("BEEPING_FTP_PASSWORD").is_err() {
            assert_eq!(backend.password, "");
        }
    }

    #[test]
    fn url_credentials_pass_through_decoded() {
        let url = Url::parse("ftp://alice:p%40ss@nas.invalid:2121/repo").unwrap();
        let backend = FtpBackend::from_url(&url).unwrap();

        assert_eq!(backend.user, "alice");
        assert_eq!(backend.password, "p@ss");
        assert_eq!(backend.port, 2121);
        assert_eq!(backend.root, "/repo");
    }

    /// MLSD is the one FTP listing whose sizes can be read without
    /// guessing at a server's `ls` format, and the facts it carries are
    /// unordered, case-insensitive, and open-ended.
    #[test]
    fn mlsd_lines_give_up_their_sizes() {
        let file = Listed::parse("type=file;size=1048576;modify=20260823090000; ab12cd")
            .expect("a file entry");

        assert!(matches!(file.kind, ListedKind::File));
        assert_eq!(file.size, 1048576);
        assert_eq!(file.name, "ab12cd");

        // Order and case are the server's business, and unknown facts
        // are to be ignored rather than tripped over
        let directory =
            Listed::parse("perm=fle;Type=dir;unique=12U1; chunks").expect("a directory");
        assert!(matches!(directory.kind, ListedKind::Directory));
        assert_eq!(directory.name, "chunks");

        // Some servers answer with full paths, as they do for NLST
        assert_eq!(
            Listed::parse("type=dir; /backups/repo/chunks/00")
                .expect("a directory")
                .name,
            "00"
        );

        // A name may hold spaces; only the first one ends the facts
        assert_eq!(
            Listed::parse("type=file;size=3; two words")
                .expect("a file")
                .name,
            "two words"
        );

        // This directory and its parent are listed for navigation, and
        // adding them up would count the tree twice over
        assert!(Listed::parse("type=cdir;modify=20260823090000; .").is_none());
        assert!(Listed::parse("type=pdir; ..").is_none());

        // Nothing to make of these
        assert!(Listed::parse("size=12; no-type").is_none());
        assert!(Listed::parse("").is_none());
        assert!(Listed::parse("type=file;size=7;").is_none());
    }
}