//! Stopping drains (oxygenhub#233): its own test binary, as stopping is
//! for the whole process.

use oxygenhub::config::Config;
use oxygenhub::db;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

/// A live event stream ends when the hub begins stopping, so it cannot
/// hold the drain open, and `serve` returns well within the drain's bound.
#[tokio::test(flavor = "multi_thread")]
async fn stopping_ends_event_streams_and_serve_returns() {
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("OXY_HOME", home.path());
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
    config.server.bind = format!("127.0.0.1:{port}");
    config.server.base_url = format!("http://127.0.0.1:{port}");
    let app = oxygenhub::build(config).await.unwrap();
    let repo = db::repos::create(&app.db, "andy", "demo", "", hub_api::Visibility::Public).await.unwrap();
    let server = tokio::spawn(oxygenhub::serve(app.clone()));

    // An open event stream, as a browser on a live page holds.
    let mut stream = None;
    for _ in 0..50 {
        if let Ok(s) = tokio::net::TcpStream::connect(("127.0.0.1", port)).await {
            stream = Some(s);
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let mut stream = stream.expect("the hub listens");
    let request = format!("GET /live?topics=repo:{} HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: text/event-stream\r\n\r\n", repo.id);
    stream.write_all(request.as_bytes()).await.unwrap();
    let mut head = vec![0u8; 512];
    let n = stream.read(&mut head).await.unwrap();
    let head = String::from_utf8_lossy(&head[..n]).to_string();
    assert!(head.starts_with("HTTP/1.1 200") && head.contains("text/event-stream"), "{head}");

    let started = std::time::Instant::now();
    oxygenhub::shutdown::begin();
    // The stream ends (the server closes it) rather than staying open.
    let mut rest = Vec::new();
    tokio::time::timeout(std::time::Duration::from_secs(10), stream.read_to_end(&mut rest))
        .await
        .expect("the event stream ends when stopping begins")
        .unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(10), server).await.expect("serve returns once drained").unwrap().unwrap();
    assert!(started.elapsed() < oxygenhub::shutdown::DRAIN, "drained in {:?}", started.elapsed());
    assert!(oxygenhub::shutdown::is_stopping());
}