//! 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());

}