//! GOAWAY on shutdown (`draft-ietf-moq-transport-22` §6.6.1, §9.2): every session is told,
//! with the New Session URI and the Timeout; new requests are refused with GOING_AWAY; a
//! client that leaves is let go, and one that stays is closed with GOAWAY_TIMEOUT.

mod common;

use common::*;
use joule_moq::draft22::{Parameters, RequestErrorCode, SessionError};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::TransportError;
use std::time::Duration;
use tokio::time::timeout;

const URI: &str = "moqt://127.0.0.1:4999/moq";

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn goaway_on_shutdown() {
    timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
}

async fn goaway_of(c: &Client) -> joule_moq::draft22::Goaway {
    let mut w = c.goaway();
    timeout(STEP, async {
        loop {
            if let Some(g) = w.borrow_and_update().clone() {
                return g;
            }
            w.changed().await.expect("the session lives until its GOAWAY");
        }
    })
    .await
    .expect("a GOAWAY in time")
}

async fn run() {
    let server = relay(|c| {
        c.limits.goaway_timeout_ms = 1500;
        c.goaway_uri = URI.into();
    })
    .await;
    let t = Tracks::new(ns(&["live", "g"]), "cam", 10, 4, 128);
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _pub = publish_tracks(&p, &t, None).await;
    let (a, _ea) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (b, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _sa = a.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap();
    let _sb = b.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap();

    let relay = server.relay().clone();
    let shutdown = tokio::spawn(async move { server.shutdown().await });

    for c in [&p, &a, &b] {
        let g = goaway_of(c).await;
        assert_eq!(g.new_session_uri, URI.as_bytes(), "the New Session URI");
        assert_eq!(g.timeout_ms, 1500, "the Timeout the relay will wait");
    }
    // A new request after GOAWAY is refused (§9.2: "MAY reject new requests").
    let refused = b.subscribe(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.expect_err("a new SUBSCRIBE after GOAWAY");
    assert!(refused.contains(&RequestErrorCode::GOING_AWAY.to_string()), "{refused}");

    // The publisher and one subscriber migrate and close with NO_ERROR; the other stays.
    a.close(SessionError::NO_ERROR.0, "migrated");
    p.close(SessionError::NO_ERROR.0, "migrated");
    match timeout(STEP, b.conn.closed()).await.expect("closed in time") {
        TransportError::Closed { code, by_peer, .. } => {
            assert_eq!(code, SessionError::GOAWAY_TIMEOUT.0, "the relay closes a session that stayed with GOAWAY_TIMEOUT");
            assert!(by_peer);
        }
        other => panic!("{other:?}"),
    }
    let closed = timeout(STEP, shutdown).await.expect("the shutdown ends").unwrap();
    assert_eq!(closed, 1, "one session outlived the GOAWAY timeout");
    assert!(until(STEP, || relay.summaries().len() == 3).await);
    let codes: Vec<Option<u64>> = relay.summaries().iter().map(|s| s.summary.close_code).collect();
    assert_eq!(codes.iter().filter(|c| **c == Some(SessionError::GOAWAY_TIMEOUT.0)).count(), 1, "{codes:?}");
    assert_eq!(codes.iter().filter(|c| **c == Some(SessionError::NO_ERROR.0)).count(), 2, "{codes:?}");
}