//! Anonymous FTP transport for the Launchpad PPA upload queue: the graceful
//! degradation of the SFTP transport when the SSH connection itself never
//! comes up (name resolution, TCP, banner or key exchange). dput-ng's plain
//! `ppa:user/ppa` profile pushes over this same queue — ppa.launchpad.net
//! over FTP, anonymous login, incoming `~user/ppa` — so it is the
//! interoperability-tested path.
//!
//! The client is [`suppaftp`]'s (plain-FTP, no TLS) blocking stream on the
//! same time-bounded sockets as the ssh2 transport: the TCP connect and
//! the control/data channel reads and writes all carry timeouts, so a
//! black-holed or stalled server fails the upload instead of hanging it
//! (suppaftp's defaults do not bound them). The upload order (payload
//! first, `.changes` last — the caller passes the files in that order) and
//! the best-effort cleanup of a failed upload (`DELE` of what was already
//! pushed, in reverse upload order) mirror the SFTP path exactly.

use std::fs::File;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::path::{Path, PathBuf};
use std::time::Duration;

use suppaftp::types::FileType;
use suppaftp::{FtpError, FtpStream};

use super::ssh;

/// Bounds one TCP connection attempt, to the control channel or a passive
/// data port: `connect(2)` would otherwise block for minutes (or forever,
/// behind a silent firewall). Generous enough for slow links to Launchpad,
/// short enough that a dead target fails in seconds — the same value and
/// rationale as the SSH path's bound.
const TCP_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);

/// Read/write timeout on the control channel: every reply is a few bytes,
/// so a stalled server has had its say by the time this expires.
const CONTROL_TIMEOUT: Duration = Duration::from_secs(60);

/// Write timeout of one data-transfer chunk: the clock restarts at every
/// write, so the wall-clock duration of a large upload is not bounded by
/// design — but a black-holed data connection fails one chunk after this
/// budget.
const DATA_TIMEOUT: Duration = Duration::from_secs(120);

/// Data transfers stream in chunks of this size.
const CHUNK_SIZE: usize = 32 * 1024;

/// Upload `files` to the `incoming` queue of `host:port` over anonymous FTP,
/// in the order given (payload first, `.changes` last — the queue processor
/// must never observe a `.changes` without its payload).
/// `on_progress(name, uploaded_bytes, total_bytes)` reports progress.
///
/// A failure mid-upload best-effort removes what was already pushed (`DELE`,
/// reverse upload order — a lingering payload in the write-only queue area
/// is only hygiene) before returning the original error, like the SFTP
/// path's [`super::cleanup_partial_upload`].
pub fn upload_queue(
    host: &str,
    port: u16,
    incoming: &str,
    files: &[(PathBuf, String)],
    on_progress: &dyn Fn(&str, u64, u64),
) -> Result<(), Box<dyn std::error::Error>> {
    let mut ftp = connect(host, port)?;
    let who = std::env::var("USER").unwrap_or_else(|_| "anonymous".to_string());
    ftp.login("anonymous".to_string(), format!("{who}@pkh.invalid"))
        .map_err(|e| format!("the FTP queue rejected the anonymous login: {e}"))?;
    ftp.cwd(incoming)
        .map_err(|e| format!("cannot enter the upload queue '{incoming}': {e}"))?;
    ftp.transfer_type(FileType::Binary)
        .map_err(|e| format!("the FTP queue refused binary transfers: {e}"))?;

    // Remote names pushed so far, in upload order, for the cleanup
    let mut uploaded: Vec<String> = Vec::new();
    for (path, name) in files {
        if let Err(e) = store(&mut ftp, path, name, on_progress) {
            // The failed file itself joins the cleanup: its `STOR` was
            // accepted before the transfer failure, so a partial may be
            // sitting in the queue
            for leftover in super::cleanup_list(&uploaded, Some(name)) {
                match ftp.rm(leftover.as_str()) {
                    Ok(()) => {
                        log::info!("Removed leftover {leftover} from the failed upload")
                    }
                    Err(e) => log::warn!(
                        "Could not remove the leftover {leftover} of the \
                         failed upload: {e}"
                    ),
                }
            }
            return Err(e);
        }
        uploaded.push(name.clone());
    }

    // Best effort: the queue keeps what it accepted, so a failure here must
    // not fail a completed upload
    if let Err(e) = ftp.quit() {
        log::debug!("closing the FTP session: {e}");
    }
    Ok(())
}

/// Connect to the queue, read its banner and set every time bound and
/// workaround the plain suppaftp stream does not carry by itself.
fn connect(host: &str, port: u16) -> Result<FtpStream, Box<dyn std::error::Error>> {
    let tcp = ssh::tcp_connect(host, port)?;
    let mut ftp = FtpStream::connect_with_stream(tcp)?.passive_stream_builder(data_connect);
    {
        let control = ftp.get_ref();
        control.set_read_timeout(Some(CONTROL_TIMEOUT))?;
        control.set_write_timeout(Some(CONTROL_TIMEOUT))?;
    }
    // A `PASV` reply announcing an unroutable address (a server behind NAT
    // that does not know its public IP) means the control connection's peer
    ftp.set_passive_nat_workaround(true);
    Ok(ftp)
}

/// The passive data-channel connect: bounded like every other network call,
/// where suppaftp's default builder is a plain, unbounded
/// `TcpStream::connect`.
fn data_connect(addr: SocketAddr) -> Result<TcpStream, FtpError> {
    TcpStream::connect_timeout(&addr, TCP_CONNECT_TIMEOUT).map_err(FtpError::ConnectionError)
}

/// Upload `path` as `name` over a passive data connection, reporting
/// progress through `on_progress` per chunk. `finish` reads the transfer
/// completion reply — the only way to learn the server accepted the file.
fn store(
    ftp: &mut FtpStream,
    path: &Path,
    name: &str,
    on_progress: &dyn Fn(&str, u64, u64),
) -> Result<(), Box<dyn std::error::Error>> {
    let mut upload = ftp
        .put_with_stream(name)
        .map_err(|e| format!("the FTP queue rejected '{name}': {e}"))?;
    if let suppaftp::DataStream::Tcp(socket) = upload.get_mut() {
        socket.set_write_timeout(Some(DATA_TIMEOUT))?;
    }

    let mut file =
        File::open(path).map_err(|e| format!("cannot read '{}': {e}", path.display()))?;
    let total = file.metadata().map(|m| m.len()).unwrap_or(0);
    let mut buffer = vec![0u8; CHUNK_SIZE];
    let mut uploaded: u64 = 0;
    loop {
        let read = file.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        upload.write_all(&buffer[..read])?;
        uploaded += read as u64;
        on_progress(name, uploaded, total);
    }
    upload
        .finish()
        .map_err(|e| format!("upload of '{name}' failed: {e}"))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use std::net::TcpListener;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use parking_lot::Mutex;

    /// In-process fake of the Launchpad FTP upload queue: one control
    /// session speaking the protocol subset the client uses (banner,
    /// USER/PASS, CWD, TYPE, PASV, STOR, DELE, QUIT). Records every
    /// command, stores received bytes under the `STOR` name, and can be
    /// told to reject one `STOR` (by index) to exercise the cleanup. Its
    /// `PASV` replies announce `0.0.0.0`, the NAT form, so the
    /// happy-path test proves the control-peer fallback too.
    struct FakeQueue {
        addr: SocketAddr,
        commands: Arc<Mutex<Vec<String>>>,
        files: Arc<Mutex<HashMap<String, Vec<u8>>>>,
    }

    impl FakeQueue {
        /// Serve one session on a background thread, greeting it with
        /// `banner` and rejecting the `STOR` number `reject_stor`, when
        /// set.
        fn start(banner: &str, reject_stor: Option<usize>) -> FakeQueue {
            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
            let addr = listener.local_addr().unwrap();
            let commands = Arc::new(Mutex::new(Vec::new()));
            let files = Arc::new(Mutex::new(HashMap::new()));
            let (c, f) = (Arc::clone(&commands), Arc::clone(&files));
            let banner = banner.to_string();
            std::thread::spawn(move || serve(listener, &banner, c, f, reject_stor));
            FakeQueue {
                addr,
                commands,
                files,
            }
        }
    }

    fn serve(
        listener: TcpListener,
        banner: &str,
        commands: Arc<Mutex<Vec<String>>>,
        files: Arc<Mutex<HashMap<String, Vec<u8>>>>,
        reject_stor: Option<usize>,
    ) {
        use std::io::BufRead;

        let (stream, _) = listener.accept().unwrap();
        let mut reader = std::io::BufReader::new(stream.try_clone().unwrap());
        let mut writer = stream;
        writer
            .write_all(format!("{banner}\r\n").as_bytes())
            .unwrap();

        let mut pending_data: Option<TcpListener> = None;
        let stor_index = AtomicUsize::new(0);
        loop {
            let mut line = String::new();
            if reader.read_line(&mut line).unwrap_or(0) == 0 {
                break;
            }
            let cmd = line.trim_end().to_string();
            commands.lock().push(cmd.clone());
            let (verb, arg) = cmd.split_once(' ').unwrap_or((cmd.as_str(), ""));
            match verb {
                "USER" | "PASS" => {
                    writer
                        .write_all(b"230 Anonymous login ok, access restrictions apply.\r\n")
                        .unwrap();
                }
                "CWD" => {
                    writer.write_all(b"250 Command successful\r\n").unwrap();
                }
                "TYPE" => {
                    writer.write_all(b"200 Type set to I\r\n").unwrap();
                }
                "PASV" => {
                    let data = TcpListener::bind("127.0.0.1:0").unwrap();
                    let port = data.local_addr().unwrap().port();
                    writer
                        .write_all(
                            format!(
                                "227 Entering Passive Mode (0,0,0,0,{},{})\r\n",
                                port / 256,
                                port % 256
                            )
                            .as_bytes(),
                        )
                        .unwrap();
                    pending_data = Some(data);
                }
                "STOR" => {
                    let index = stor_index.fetch_add(1, Ordering::SeqCst);
                    if Some(index) == reject_stor {
                        writer.write_all(b"550 Permission denied\r\n").unwrap();
                        continue;
                    }
                    writer.write_all(b"150 Ok to send data\r\n").unwrap();
                    let (mut data, _) = pending_data.take().unwrap().accept().unwrap();
                    let mut bytes = Vec::new();
                    data.read_to_end(&mut bytes).unwrap();
                    files.lock().insert(arg.to_string(), bytes);
                    writer.write_all(b"226 Transfer complete\r\n").unwrap();
                }
                "DELE" => {
                    if files.lock().remove(arg).is_some() {
                        writer.write_all(b"250 File deleted\r\n").unwrap();
                    } else {
                        writer.write_all(b"550 No such file\r\n").unwrap();
                    }
                }
                "QUIT" => {
                    writer.write_all(b"221 Bye\r\n").unwrap();
                    break;
                }
                other => {
                    writer
                        .write_all(format!("502 Command '{other}' not implemented\r\n").as_bytes())
                        .unwrap();
                }
            }
        }
    }

    /// A temp file with `content`, to upload.
    fn upload_file(dir: &Path, name: &str, content: &[u8]) -> (PathBuf, String) {
        let path = dir.join(name);
        std::fs::write(&path, content).unwrap();
        (path, name.to_string())
    }

    /// The full push: anonymous login, the queue directory entered,
    /// payload files uploaded in order before the `.changes`, their bytes
    /// intact, and progress reported up to each file's size.
    #[test]
    fn upload_queue_pushes_payload_before_changes() {
        let queue = FakeQueue::start("220 Launchpad upload server", None);
        let dir = tempfile::tempdir().unwrap();
        let files = vec![
            upload_file(dir.path(), "pkg_1.0.orig.tar.xz", b"orig bytes"),
            upload_file(dir.path(), "pkg_1.0-1.dsc", b"dsc bytes"),
            upload_file(dir.path(), "pkg_1.0-1_source.changes", b"changes bytes"),
        ];
        let seen = Arc::new(Mutex::new(Vec::new()));
        let progress_log = Arc::clone(&seen);

        upload_queue(
            "127.0.0.1",
            queue.addr.port(),
            "~vhaudiquet/lp2167827",
            &files,
            &|name, uploaded, total| {
                progress_log
                    .lock()
                    .push((name.to_string(), uploaded, total))
            },
        )
        .unwrap();

        let commands = queue.commands.lock().clone();
        assert_eq!(
            commands[0], "USER anonymous",
            "the anonymous login comes first"
        );
        assert_eq!(
            commands
                .iter()
                .filter(|c| c == &"CWD ~vhaudiquet/lp2167827")
                .count(),
            1
        );
        let cwd = commands.iter().position(|c| c.starts_with("CWD")).unwrap();
        let first_stor = commands.iter().position(|c| c.starts_with("STOR")).unwrap();
        assert!(cwd < first_stor, "the queue directory is entered first");
        assert!(
            commands.iter().any(|c| c == "TYPE I"),
            "binary transfers are requested"
        );
        // Upload order: the payload first, the .changes last
        let stors: Vec<&String> = commands.iter().filter(|c| c.starts_with("STOR")).collect();
        assert_eq!(
            stors.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
            vec![
                "STOR pkg_1.0.orig.tar.xz",
                "STOR pkg_1.0-1.dsc",
                "STOR pkg_1.0-1_source.changes",
            ]
        );
        assert_eq!(*commands.last().unwrap(), "QUIT");

        // The received bytes are intact
        let stored = queue.files.lock();
        assert_eq!(stored.get("pkg_1.0.orig.tar.xz").unwrap(), b"orig bytes");
        assert_eq!(stored.get("pkg_1.0-1.dsc").unwrap(), b"dsc bytes");
        assert_eq!(
            stored.get("pkg_1.0-1_source.changes").unwrap(),
            b"changes bytes"
        );

        // Progress reached each file's size
        let progress = seen.lock();
        for (path, name) in &files {
            let size = path.metadata().unwrap().len();
            let reached = progress
                .iter()
                .any(|(n, uploaded, total)| n == name && *uploaded == size && *total == size);
            assert!(reached, "no progress report completed for '{name}'");
        }
    }

    /// A `STOR` the queue rejects fails the upload, and the client removes
    /// what it already pushed — the failed file first, then the earlier
    /// payloads, in reverse upload order — before returning the error.
    #[test]
    fn upload_queue_cleans_up_after_a_rejected_stor() {
        let queue = FakeQueue::start("220 Launchpad upload server", Some(1));
        let dir = tempfile::tempdir().unwrap();
        let files = vec![
            upload_file(dir.path(), "one.dsc", b"one"),
            upload_file(dir.path(), "two.tar.xz", b"two"),
            upload_file(dir.path(), "three.changes", b"three"),
        ];

        let err = upload_queue(
            "127.0.0.1",
            queue.addr.port(),
            "~user/ppa",
            &files,
            &|_, _, _| {},
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("rejected 'two.tar.xz'"),
            "the error names the rejected file, got: {err}"
        );

        let commands = queue.commands.lock();
        let deles: Vec<&String> = commands.iter().filter(|c| c.starts_with("DELE")).collect();
        assert_eq!(
            deles.iter().map(|d| d.as_str()).collect::<Vec<_>>(),
            vec!["DELE two.tar.xz", "DELE one.dsc"],
            "the failed file is removed first, then the earlier payloads"
        );
        let files = queue.files.lock();
        assert!(
            !files.contains_key("one.dsc"),
            "the pushed payload is removed"
        );
        assert!(!files.contains_key("three.changes"), "never reached");
    }

    /// A multi-line banner (`220-first` closed by `220 last`) parses like
    /// the single-line form.
    #[test]
    fn upload_queue_reads_multiline_replies() {
        let queue = FakeQueue::start("220-Launchpad\r\n220 upload server", None);
        let dir = tempfile::tempdir().unwrap();
        let files = vec![upload_file(dir.path(), "pkg.dsc", b"bytes")];

        upload_queue(
            "127.0.0.1",
            queue.addr.port(),
            "~user/ppa",
            &files,
            &|_, _, _| {},
        )
        .unwrap();

        let stored = queue.files.lock();
        assert_eq!(stored.get("pkg.dsc").unwrap(), b"bytes");
    }

    /// Live control-channel handshake with the real Launchpad FTP queue:
    /// banner, anonymous login and a `CWD` — nothing is uploaded, the
    /// queue is left untouched. For deliberate ad-hoc runs
    /// (`cargo test -- --ignored`), not the pre-commit pass: it hits the
    /// network.
    #[test]
    #[ignore = "hits the network: the real Launchpad FTP queue"]
    fn live_launchpad_control_channel() {
        let (host, port) = crate::launchpad::ppa_ftp_queue();
        assert_eq!((host.as_str(), port), ("ppa.launchpad.net", 21));

        let mut ftp = connect(&host, port).unwrap();
        ftp.login("anonymous", "pkh@invalid").unwrap();
        ftp.cwd("/").unwrap();
        ftp.quit().unwrap();
    }
}