//! Request IDs and requests in flight (`draft-ietf-moq-transport-22` §6.4.2): a Request ID
//! of the server's parity, or one seen before, closes the session with INVALID_REQUEST_ID;
//! and the requests a client may have open at once are bounded by the relay's stream limit
//! (here WT_MAX_STREAMS on the WebSocket binding), credit returning as requests end.

mod common;

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

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

fn track_status(id: u64) -> ControlMessage {
    ControlMessage::TrackStatus(TrackRequest { request_id: id, namespace: ns(&["x"]), name: name("y"), params: Parameters::new() })
}

async fn closed_with(c: &joule_moq_relay::client::Client) -> u64 {
    match timeout(STEP, c.conn.closed()).await.expect("closed in time") {
        TransportError::Closed { code, .. } => code,
        other => panic!("{other:?}"),
    }
}

async fn run() {
    let server = relay(|c| c.limits.max_requests = 2).await;

    // A Request ID with the server's parity (odd).
    let (a, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _r = a.request(&track_status(1)).await.unwrap();
    assert_eq!(closed_with(&a).await, SessionError::INVALID_REQUEST_ID.0);

    // A Request ID used twice.
    let (b, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let (_t1, mut r1) = b.request(&track_status(0)).await.unwrap();
    let _ = joule_moq_relay::client::Client::response(&mut r1).await;
    let _r2 = b.request(&track_status(0)).await.unwrap();
    assert_eq!(closed_with(&b).await, SessionError::INVALID_REQUEST_ID.0);

    // Two requests open at once; a third waits for credit until one of them ends.
    let t = Tracks::new(ns(&["live", "r"]), "cam", 10, 6, 128);
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _pub = publish_tracks(&p, &t, None).await;
    let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let (_ok1, tx1, _rx1) = c.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap();
    let (_ok2, _tx2, _rx2) = c.subscribe(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.unwrap();
    assert!(timeout(Duration::from_millis(400), c.conn.open_bi()).await.is_err(), "a third request opened past the stream limit");
    // Cancelling the first returns its credit (§6.4.2.3).
    tx1.reset(StreamResetCode::CANCELLED.0).await;
    let third = timeout(STEP, c.conn.open_bi()).await;
    assert!(matches!(third, Ok(Ok(_))), "the third request opened once the first ended");

    c.close(0, "done");
    p.close(0, "done");
    server.shutdown().await;
}