use std::path::Path;
use std::process::{Child, Command, Output, Stdio};
use std::time::{Duration, Instant};

/// a hydrant on a database in `dir` that never touches the network
fn offline_hydrant(dir: &Path) -> Command {
    offline(Command::new(env!("CARGO_BIN_EXE_hydrant")), dir)
}

fn offline(mut cmd: Command, dir: &Path) -> Command {
    cmd.env("HYDRANT_DATABASE_PATH", dir.join("db"))
        .env(
            "HYDRANT_API_BIND",
            format!("unix:{}", dir.join("api.sock").display()),
        )
        .env("HYDRANT_ENABLE_FIREHOSE", "false")
        .env("HYDRANT_ENABLE_CRAWLER", "false")
        .env("HYDRANT_ENABLE_BACKFILL", "false")
        .env("HYDRANT_RELAY_HOSTS", "")
        .env("HYDRANT_SEED_HOSTS", "")
        .env("HYDRANT_CRAWLER_URLS", "")
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    cmd
}

// config is env only, so anything on argv has to be refused before hydrant boots.
// if it does boot, it gets a throwaway database and no network, and we kill it.
fn hydrant(args: &[&str]) -> Output {
    let tmp = tempfile::tempdir().unwrap();
    let mut child = offline_hydrant(tmp.path()).args(args).spawn().unwrap();
    let deadline = Instant::now() + Duration::from_secs(30);
    while child.try_wait().unwrap().is_none() {
        if Instant::now() > deadline {
            child.kill().unwrap();
            let out = child.wait_with_output().unwrap();
            panic!(
                "hydrant {args:?} booted instead of exiting:\n{}",
                String::from_utf8_lossy(&out.stderr)
            );
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    child.wait_with_output().unwrap()
}

#[test]
fn version_prints_and_exits() {
    for flag in ["--version", "-V"] {
        let out = hydrant(&[flag]);
        assert!(out.status.success(), "{flag}: {out:?}");
        assert_eq!(
            String::from_utf8_lossy(&out.stdout),
            format!("hydrant {}\n", env!("CARGO_PKG_VERSION"))
        );
    }
}

#[test]
fn help_prints_usage_and_exits() {
    for flag in ["--help", "-h"] {
        let out = hydrant(&[flag]);
        assert!(out.status.success(), "{flag}: {out:?}");
        assert!(String::from_utf8_lossy(&out.stdout).starts_with("usage: hydrant"));
    }
}

#[test]
fn any_other_argument_is_an_error() {
    for args in [&["--port=3000"][..], &["serve"], &["--version", "extra"]] {
        let out = hydrant(args);
        assert!(!out.status.success(), "{args:?}: {out:?}");
        assert!(out.stdout.is_empty(), "{args:?}: {out:?}");
        assert!(
            String::from_utf8_lossy(&out.stderr).contains("unexpected argument"),
            "{args:?}: {out:?}"
        );
    }
}

/// waits for `child` to exit on its own, killing it after `timeout`
fn wait_for_exit(mut child: Child, timeout: Duration) -> Option<Output> {
    let deadline = Instant::now() + timeout;
    while child.try_wait().unwrap().is_none() {
        if Instant::now() > deadline {
            child.kill().unwrap();
            return None;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    Some(child.wait_with_output().unwrap())
}

fn send_signal(child: &Child, signal: &str) {
    let status = Command::new("kill")
        .args([signal, &child.id().to_string()])
        .status()
        .unwrap();
    assert!(status.success(), "kill {signal} failed");
}

/// starts hydrant on `dir` and returns once its api socket is up, so it's fully running
fn start_and_wait_for_api(dir: &Path) -> Child {
    let socket = dir.join("api.sock");
    let _ = std::fs::remove_file(&socket);
    let mut child = offline_hydrant(dir).spawn().unwrap();
    let deadline = Instant::now() + Duration::from_secs(60);
    while !socket.exists() {
        if let Some(status) = child.try_wait().unwrap() {
            let out = child.wait_with_output().unwrap();
            panic!(
                "hydrant exited with {status} before serving:\n{}",
                String::from_utf8_lossy(&out.stderr)
            );
        }
        assert!(
            Instant::now() < deadline,
            "hydrant never bound its api socket"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
    child
}

#[test]
fn sigterm_and_sigint_stop_cleanly_and_the_database_opens_again() {
    let tmp = tempfile::tempdir().unwrap();
    // the second and third start only get past opening the database if the one before let go
    for signal in ["-TERM", "-INT", "-TERM"] {
        let child = start_and_wait_for_api(tmp.path());
        send_signal(&child, signal);
        let out = wait_for_exit(child, Duration::from_secs(45))
            .unwrap_or_else(|| panic!("hydrant didn't stop within 45s of {signal}"));
        let stdout = String::from_utf8_lossy(&out.stdout);
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            out.status.success(),
            "{signal}: {:?}\n{stdout}{stderr}",
            out.status
        );
        assert!(stdout.contains("stopped"), "{signal}: {stdout}{stderr}");
    }
}

#[test]
fn config_warnings_are_logged() {
    let tmp = tempfile::tempdir().unwrap();
    // the bad history ttl makes it exit right after reading the config
    let child = offline_hydrant(tmp.path())
        .env("HYDRANT_RELAY_HOSTS", "not a url")
        .env("HYDRANT_HISTORY_TTL", "soon")
        .spawn()
        .unwrap();
    let out = wait_for_exit(child, Duration::from_secs(30)).expect("hydrant booted");
    assert!(!out.status.success(), "{out:?}");
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        stdout.contains("invalid relay host URL: not a url"),
        "{stdout}{}",
        String::from_utf8_lossy(&out.stderr)
    );
}

#[cfg(feature = "indexer_stream")]
fn socket_activated(dir: &Path, listener: &std::net::TcpListener) -> Child {
    use std::os::fd::AsRawFd;
    use std::os::unix::process::CommandExt;

    // exec keeps the shell's pid, which makes $$ hydrant's pid
    let mut sh = Command::new("sh");
    sh.args(["-c", "LISTEN_PID=$$ exec \"$0\"", env!("CARGO_BIN_EXE_hydrant")]);
    let mut cmd = offline(sh, dir);
    cmd.env("HYDRANT_API_BIND", listener.local_addr().unwrap().to_string())
        .env("LISTEN_FDS", "1");
    let fd = listener.as_raw_fd();
    // safety: dup2 and fcntl are async-signal-safe
    unsafe {
        cmd.pre_exec(move || {
            if fd == 3 {
                libc::fcntl(3, libc::F_SETFD, 0);
            } else if libc::dup2(fd, 3) < 0 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    cmd.spawn().unwrap()
}

/// connect succeeds off the backlog and only the read waits
#[cfg(feature = "indexer_stream")]
fn get_health(addr: std::net::SocketAddr) -> String {
    use std::io::{Read, Write};
    let mut conn = std::net::TcpStream::connect(addr).expect("the listener is never closed");
    conn.set_read_timeout(Some(Duration::from_secs(60))).unwrap();
    conn.write_all(b"GET /health HTTP/1.1\r\nhost: hydrant\r\nconnection: close\r\n\r\n")
        .unwrap();
    let mut response = String::new();
    conn.read_to_string(&mut response).unwrap();
    response
}

#[cfg(feature = "indexer_stream")]
#[test]
fn a_socket_activated_restart_closes_streams_and_keeps_new_clients_waiting() {
    use futures::StreamExt;
    use tokio_websockets::{ClientBuilder, CloseCode};

    let tmp = tempfile::tempdir().unwrap();
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let first = socket_activated(tmp.path(), &listener);
    assert!(get_health(addr).starts_with("HTTP/1.1 200"));
    let uri = format!("ws://{addr}/stream");
    let (mut subscriber, _) = rt
        .block_on(ClientBuilder::new().uri(&uri).unwrap().connect())
        .unwrap();

    send_signal(&first, "-TERM");
    let closed = rt
        .block_on(async { tokio::time::timeout(Duration::from_secs(10), subscriber.next()).await })
        .expect("the stream must be closed when hydrant stops")
        .expect("a close frame, not the socket just going")
        .unwrap();
    assert_eq!(
        closed.as_close().map(|(code, _)| code),
        Some(CloseCode::SERVICE_RESTART)
    );
    let out = wait_for_exit(first, Duration::from_secs(45)).expect("hydrant didn't stop");
    assert!(out.status.success(), "{out:?}");

    let waiting = std::thread::spawn(move || get_health(addr));
    std::thread::sleep(Duration::from_millis(200));
    let second = socket_activated(tmp.path(), &listener);
    let response = waiting.join().unwrap();
    assert!(response.starts_with("HTTP/1.1 200"), "{response}");

    send_signal(&second, "-TERM");
    let out = wait_for_exit(second, Duration::from_secs(45)).expect("hydrant didn't stop");
    assert!(out.status.success(), "{out:?}");
}