use futures::{Sink, Stream};
use qmux::tungstenite;
use std::{
	pin::Pin,
	sync::{Arc, atomic::Ordering},
	task::{Context, Poll},
};

use axum::{
	extract::{Extension, OriginalUri, State, WebSocketUpgrade, ws::rejection::WebSocketUpgradeRejection},
	http::{HeaderMap, HeaderValue, StatusCode, Uri, header::HOST},
	response::Response,
};
use moq_net::origin;
use moq_net::stats::Session;

use crate::{Auth, AuthParams, web::MtlsPeer, web::WebState, web::landing_response};

pub(crate) async fn serve_ws(
	ws: Result<WebSocketUpgrade, WebSocketUpgradeRejection>,
	OriginalUri(uri): OriginalUri,
	headers: HeaderMap,
	mtls: Option<Extension<MtlsPeer>>,
	State(state): State<Arc<WebState>>,
) -> axum::response::Result<Response> {
	// If this isn't a WebSocket upgrade (e.g. a plain browser visit), serve
	// the informational landing page instead of an error response.
	let Ok(ws) = ws else {
		return Ok(landing_response());
	};

	let ws = negotiate_subprotocol(ws)?;

	let host = uri
		.authority()
		.map(axum::http::uri::Authority::as_str)
		.or_else(|| headers.get(HOST).and_then(|value| value.to_str().ok()))
		.ok_or(StatusCode::BAD_REQUEST)?;
	let mut params = request_auth_params(&state.auth, host, &uri)?;
	params.transport = Some(moq_native::Transport::WebSocket);
	let token = if mtls.is_some() {
		// mTLS peers: the API returns the canonical root and the billing tier.
		state.auth.verify_mtls(&params.path, params.transport).await?
	} else {
		state.auth.verify(&params).await?
	};
	let publish = state.cluster.publisher(&token);
	let subscribe = state.cluster.subscriber(&token);
	let stats = state.cluster.stats.tier(token.tier.clone()).session(&token.root);

	if publish.is_none() && subscribe.is_none() {
		// Bad token, we can't publish or subscribe.
		return Err(StatusCode::UNAUTHORIZED.into());
	}

	Ok(ws.on_upgrade(async move |socket| {
		let id = state.conn_id.fetch_add(1, Ordering::Relaxed);

		// Capture the negotiated subprotocol before we erase the WebSocket type
		// in the Stream/Sink adapters; qmux needs it to derive the moq version.
		let alpn = socket.protocol().and_then(|h| h.to_str().ok()).map(str::to_owned);

		// Unfortunately, we need to convert from Axum to Tungstenite.
		// Axum uses Tungstenite internally, but it's not exposed to avoid semvar issues.
		let socket = WebSocketAdapter::new(socket);
		let _ = handle_socket(id, socket, alpn, publish, subscribe, stats).await;
	}))
}

/// Apply the same host and path authentication routing used by native WebTransport.
fn request_auth_params(auth: &Auth, host: &str, uri: &Uri) -> Result<AuthParams, StatusCode> {
	let path = uri.path_and_query().ok_or(StatusCode::BAD_REQUEST)?;
	let url = url::Url::parse(&format!("https://{host}{path}")).map_err(|_| StatusCode::BAD_REQUEST)?;
	Ok(auth.params_from_url(&url))
}

#[tracing::instrument("ws", err, skip_all, fields(id = _id))]
async fn handle_socket<T>(
	_id: u64,
	socket: T,
	alpn: Option<String>,
	publish: Option<origin::Producer>,
	subscribe: Option<origin::Producer>,
	stats: Session,
) -> anyhow::Result<()>
where
	T: futures::Stream<Item = Result<tungstenite::Message, tungstenite::Error>>
		+ futures::Sink<tungstenite::Message, Error = tungstenite::Error>
		+ Send
		+ Unpin
		+ 'static,
{
	// Wrap the WebSocket in a WebTransport compatibility layer. We have to
	// forward the negotiated subprotocol explicitly; axum performed the
	// upgrade, so qmux can't sniff it from the handshake.
	//
	// Keep-alive is not optional here. A peer whose host crashes or whose
	// network drops sends no FIN, so without a Ping/timeout the session stays
	// open until OS-level TCP keep-alive probes it, typically hours. Every
	// broadcast it published stays announced for that entire window and the
	// announce propagates to the rest of the cluster. QUIC gets this from its
	// idle timeout; WebSocket has no equivalent of its own.
	let upgraded = qmux::ws::Upgraded::new(socket).with_keep_alive(qmux::KeepAlive::default());
	let upgraded = match alpn.as_deref() {
		Some(alpn) => upgraded.with_alpn(alpn),
		None => upgraded,
	};
	let ws = upgraded.accept();
	// Only set the side the token actually grants. moq-net defaults the
	// unset side to a fresh no-op origin, which is fine for a
	// publish-only or subscribe-only token.
	let mut server = moq_net::Server::new().with_stats(stats);
	if let Some(subscribe) = subscribe {
		server = server.with_publisher(&subscribe);
	}
	if let Some(publish) = publish {
		server = server.with_subscriber(publish);
	}
	// Hold the session so it doesn't close early; the driver serves it in place.
	let (_session, driver) = server.accept(ws).await?;
	driver.await.map_err(Into::into)
}

/// Pick a subprotocol for the upgrade, or fail the handshake outright.
///
/// We advertise the full qmux × moq-net subprotocol matrix, with bare qmux
/// fallbacks last. axum picks the first entry that the client also offered, so
/// a modern client lands on `qmux-01.moq-lite-05`; old clients still match
/// `webtransport` or `qmux-00.moql` and negotiate via SETUP.
///
/// When the client offered subprotocols and none of them are ours, the
/// qmux-over-WebSocket draft requires us to fail the handshake rather than
/// upgrade with no selection: the client MUST treat a missing
/// `Sec-WebSocket-Protocol` response header as a failure anyway, so upgrading
/// only wastes a connection that has already agreed on nothing.
///
/// A client that offers no subprotocol at all is left alone: it upgrades and
/// negotiates the moq version over moq-lite SETUP instead.
fn negotiate_subprotocol(ws: WebSocketUpgrade) -> Result<WebSocketUpgrade, StatusCode> {
	let supported = supported_subprotocols();

	if !subprotocols_acceptable(ws.requested_protocols().map(HeaderValue::as_bytes), &supported) {
		tracing::debug!("rejecting WebSocket upgrade: no supported subprotocol offered");
		return Err(StatusCode::BAD_REQUEST);
	}

	Ok(ws.protocols(supported))
}

/// Whether the offered subprotocols leave us something to select.
///
/// True when the client offered one we support, or offered none at all. Empty
/// entries are ignored so a blank header reads as "offered nothing" instead of
/// as an identifier we don't know.
fn subprotocols_acceptable<'a>(requested: impl IntoIterator<Item = &'a [u8]>, supported: &[String]) -> bool {
	let mut offered = false;

	for protocol in requested {
		let protocol = protocol.trim_ascii();
		if protocol.is_empty() {
			continue;
		}
		offered = true;

		if supported.iter().any(|s| s.as_bytes() == protocol) {
			return true;
		}
	}

	!offered
}

/// QMux wire-format versions that can ride under a `{prefix}.{alpn}` pair.
/// Newest first so axum's exact-string match picks the freshest one.
const QMUX_VERSIONS: &[qmux::Version] = &[qmux::Version::QMux01, qmux::Version::QMux00];

/// moq-transport-18 and -19 require qmux-01, so we never pair them with qmux-00.
/// Mirrors `js/net`'s `connect.ts` and moq-native's `websocket_subprotocols`.
const QMUX01_ONLY_ALPNS: &[&str] = &["moqt-18", "moqt-19"];

/// Subprotocols to advertise on the WebSocket upgrade.
///
/// Generates the cross product of [`QMUX_VERSIONS`] × `moq_net::ALPNS`, with
/// the bare qmux fallbacks (`qmux-01`, `qmux-00`, `webtransport`) appended
/// last so versioned subprotocols always win the exact-string match axum
/// performs. Without the versioned entries, axum picks bare `webtransport`,
/// qmux can't resolve a moq version from it, and the relay silently
/// downgrades clients to Lite02 via SETUP-based negotiation.
///
/// `qmux-00.moqt-1{8,9}` is excluded: moq-transport-18 and -19 require qmux-01, so
/// those pairs are illegal.
fn supported_subprotocols() -> Vec<String> {
	let mut out = Vec::with_capacity(QMUX_VERSIONS.len() * moq_net::ALPNS.len() + qmux::ALPNS.len());
	for &version in QMUX_VERSIONS {
		for &alpn in moq_net::ALPNS {
			if version == qmux::Version::QMux00 && QMUX01_ONLY_ALPNS.contains(&alpn) {
				continue;
			}
			out.push(format!("{}{alpn}", version.prefix()));
		}
	}
	for &alpn in qmux::ALPNS {
		out.push(alpn.to_string());
	}
	out
}

// https://github.com/tokio-rs/axum/discussions/848#discussioncomment-11443587

struct WebSocketAdapter<T> {
	inner: T,
}

impl<T> WebSocketAdapter<T> {
	fn new(inner: T) -> Self {
		Self { inner }
	}
}

impl<T> Stream for WebSocketAdapter<T>
where
	T: Stream<Item = Result<axum::extract::ws::Message, axum::Error>> + Unpin,
{
	type Item = Result<tungstenite::Message, tungstenite::Error>;

	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
		Pin::new(&mut self.inner)
			.poll_next(cx)
			.map(|message| message.map(|message| message.map(axum_to_tungstenite).map_err(map_axum_error)))
	}
}

impl<T> Sink<tungstenite::Message> for WebSocketAdapter<T>
where
	T: Sink<axum::extract::ws::Message, Error = axum::Error> + Unpin,
{
	type Error = tungstenite::Error;

	fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
		Pin::new(&mut self.inner).poll_ready(cx).map_err(map_axum_error)
	}

	fn start_send(mut self: Pin<&mut Self>, message: tungstenite::Message) -> Result<(), Self::Error> {
		Pin::new(&mut self.inner)
			.start_send(tungstenite_to_axum(message))
			.map_err(map_axum_error)
	}

	fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
		Pin::new(&mut self.inner).poll_flush(cx).map_err(map_axum_error)
	}

	fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
		Pin::new(&mut self.inner).poll_close(cx).map_err(map_axum_error)
	}
}

fn map_axum_error(err: axum::Error) -> tungstenite::Error {
	tracing::warn!(%err, "WebSocket error");
	tungstenite::Error::ConnectionClosed
}

fn axum_to_tungstenite(message: axum::extract::ws::Message) -> tungstenite::Message {
	match message {
		axum::extract::ws::Message::Text(text) => tungstenite::Message::Text(text.to_string().into()),
		axum::extract::ws::Message::Binary(bin) => tungstenite::Message::Binary(Vec::from(bin).into()),
		axum::extract::ws::Message::Ping(ping) => tungstenite::Message::Ping(Vec::from(ping).into()),
		axum::extract::ws::Message::Pong(pong) => tungstenite::Message::Pong(Vec::from(pong).into()),
		axum::extract::ws::Message::Close(close) => {
			tungstenite::Message::Close(close.map(|c| tungstenite::protocol::CloseFrame {
				code: c.code.into(),
				reason: c.reason.to_string().into(),
			}))
		}
	}
}

fn tungstenite_to_axum(message: tungstenite::Message) -> axum::extract::ws::Message {
	match message {
		tungstenite::Message::Text(text) => axum::extract::ws::Message::Text(text.to_string().into()),
		tungstenite::Message::Binary(bin) => axum::extract::ws::Message::Binary(Vec::from(bin).into()),
		tungstenite::Message::Ping(ping) => axum::extract::ws::Message::Ping(Vec::from(ping).into()),
		tungstenite::Message::Pong(pong) => axum::extract::ws::Message::Pong(Vec::from(pong).into()),
		tungstenite::Message::Frame(_frame) => unreachable!(),
		tungstenite::Message::Close(close) => {
			axum::extract::ws::Message::Close(close.map(|c| axum::extract::ws::CloseFrame {
				code: c.code.into(),
				reason: c.reason.to_string().into(),
			}))
		}
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use crate::AuthConfig;
	use axum::{Router, extract::WebSocketUpgrade, routing::any};
	use futures::SinkExt;
	use std::{io, sync::atomic::AtomicBool, time::Duration};
	use tokio::sync::mpsc;
	// Brings `qmux::Session::protocol` and `::closed` into scope.
	use web_transport_trait::Session as _;

	struct DoubleErrorSocket;

	impl Stream for DoubleErrorSocket {
		type Item = Result<axum::extract::ws::Message, axum::Error>;

		fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
			Poll::Pending
		}
	}

	impl Sink<axum::extract::ws::Message> for DoubleErrorSocket {
		type Error = axum::Error;

		fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
			Poll::Ready(Ok(()))
		}

		fn start_send(self: Pin<&mut Self>, _message: axum::extract::ws::Message) -> Result<(), Self::Error> {
			Err(axum::Error::new(io::Error::from(io::ErrorKind::BrokenPipe)))
		}

		fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
			Poll::Ready(Ok(()))
		}

		fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
			Poll::Ready(Err(axum::Error::new(io::Error::from(io::ErrorKind::BrokenPipe))))
		}
	}

	#[tokio::test]
	async fn websocket_adapter_maps_close_error_after_send_error() {
		let mut socket = WebSocketAdapter::new(DoubleErrorSocket);

		let send = socket.send(tungstenite::Message::Binary(Vec::new().into())).await;
		assert!(matches!(send, Err(tungstenite::Error::ConnectionClosed)));

		let close = socket.close().await;
		assert!(matches!(close, Err(tungstenite::Error::ConnectionClosed)));
	}

	#[tokio::test]
	async fn websocket_auth_applies_subdomain_routing() {
		let config: AuthConfig = serde_json::from_value(serde_json::json!({
			"domains": ["cdn.moq.pro"],
			"public": "viewer"
		}))
		.expect("parse auth config");
		let auth = Auth::new(config).await.expect("build auth");
		for uri in ["/bbb.hang?jwt=token", "https://demo.cdn.moq.pro/bbb.hang?jwt=token"] {
			let uri: Uri = uri.parse().expect("parse URI");
			let params = request_auth_params(&auth, "demo.cdn.moq.pro", &uri).expect("build auth params");

			assert_eq!(params.path, "/demo/bbb.hang");
			assert_eq!(params.jwt.as_deref(), Some("token"));
		}
	}

	/// The newest moq ALPN both sides agree on. Derived from the same source
	/// of truth that `supported_subprotocols` and `qmux::Client::with_protocols`
	/// consume, so adding a new ALPN doesn't break these tests independently
	/// of the production logic.
	fn newest_moq_alpn() -> &'static str {
		moq_net::ALPNS.first().copied().expect("moq_net::ALPNS is empty")
	}

	fn preferred_qmux_prefix() -> &'static str {
		QMUX_VERSIONS.first().expect("QMUX_VERSIONS is empty").prefix()
	}

	#[test]
	fn supported_subprotocols_lists_full_matrix() {
		// Guard the literals: they must stay the IETF draft-18/19 ALPNs
		// (wire 0xff000012 / 0xff000013).
		assert_eq!(
			QMUX01_ONLY_ALPNS
				.iter()
				.map(|&a| moq_net::Version::from_alpn(a).map(|v| v.code()))
				.collect::<Vec<_>>(),
			vec![Some(0xff000012), Some(0xff000013)]
		);

		let list = supported_subprotocols();

		// Newest moq ALPN under the preferred prefix must come first so axum
		// picks it whenever the client offers it.
		let expected_first = format!("{}{}", preferred_qmux_prefix(), newest_moq_alpn());
		assert_eq!(list.first().map(String::as_str), Some(expected_first.as_str()));

		// Every moq ALPN must appear under every qmux wire version, except the
		// illegal `qmux-00.moqt-1{8,9}` pairs (moq-transport-18/19 need qmux-01).
		for &version in QMUX_VERSIONS {
			for &alpn in moq_net::ALPNS {
				let entry = format!("{}{alpn}", version.prefix());
				if version == qmux::Version::QMux00 && QMUX01_ONLY_ALPNS.contains(&alpn) {
					assert!(!list.contains(&entry), "illegal pair {entry} must not be advertised");
					continue;
				}
				assert!(list.contains(&entry), "missing {entry}");
			}
		}

		// Bare qmux fallbacks must come after every versioned entry so they
		// only win when the client offers nothing better. The buggy
		// `["webtransport"]` advertise list would put a bare entry first and
		// silently downgrade modern clients to Lite02.
		let last_versioned_idx = list
			.iter()
			.rposition(|s| s.contains('.'))
			.expect("no versioned entries");
		for &bare in qmux::ALPNS {
			let bare_idx = list
				.iter()
				.position(|s| s == bare)
				.unwrap_or_else(|| panic!("missing bare fallback {bare}"));
			assert!(
				bare_idx > last_versioned_idx,
				"bare {bare} must come after every versioned entry, got {list:?}",
			);
		}
	}

	#[test]
	fn subprotocols_acceptable_requires_a_match_when_any_are_offered() {
		let supported = supported_subprotocols();
		let known = supported.first().expect("no supported subprotocols").clone();

		// Offering nothing is the legacy route: upgrade and negotiate via SETUP.
		assert!(subprotocols_acceptable([], &supported));
		// A blank header carries no identifier, so it reads the same way.
		assert!(subprotocols_acceptable([b"" as &[u8], b"  "], &supported));

		// Something we know, alone or among identifiers we don't.
		assert!(subprotocols_acceptable([known.as_bytes()], &supported));
		assert!(subprotocols_acceptable(
			[b"bogus-99" as &[u8], known.as_bytes()],
			&supported
		));

		// Nothing we know: the draft says fail the handshake.
		assert!(!subprotocols_acceptable([b"bogus-99" as &[u8]], &supported));
		assert!(!subprotocols_acceptable([b"bogus-99" as &[u8], b"soap"], &supported));

		// Near misses must not sneak through a prefix or substring match.
		assert!(!subprotocols_acceptable(
			[format!("{known}-next").as_bytes()],
			&supported
		));
	}

	/// Send a raw WebSocket handshake and return its HTTP status line.
	///
	/// `protocols` is the `Sec-WebSocket-Protocol` request header, omitted when
	/// `None`. Hand-rolled because a qmux client always appends the bare
	/// fallbacks we support, so it can't express "offers only what we reject".
	async fn handshake_status(addr: std::net::SocketAddr, protocols: Option<&str>) -> String {
		use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

		let mut request = format!(
			"GET / HTTP/1.1\r\n\
			 Host: {addr}\r\n\
			 Upgrade: websocket\r\n\
			 Connection: Upgrade\r\n\
			 Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
			 Sec-WebSocket-Version: 13\r\n"
		);
		if let Some(protocols) = protocols {
			request.push_str(&format!("Sec-WebSocket-Protocol: {protocols}\r\n"));
		}
		request.push_str("\r\n");

		let mut stream = tokio::net::TcpStream::connect(addr).await.expect("connect");
		stream.write_all(request.as_bytes()).await.expect("write request");

		// Read through the newline rather than taking whatever one read returns:
		// TCP is free to split the status line across segments. We never read
		// past it, so the headers and body stay in the socket.
		let mut stream = BufReader::new(stream);
		let mut status = Vec::new();
		tokio::time::timeout(Duration::from_secs(5), stream.read_until(b'\n', &mut status))
			.await
			.expect("server did not respond")
			.expect("read response");

		String::from_utf8_lossy(&status).trim_end().to_owned()
	}

	/// A client offering only subprotocols we don't support must fail the
	/// handshake, per draft-lcurley-qmux-websocket. Upgrading anyway leaves the
	/// connection with no negotiated ALPN, which the client is required to treat
	/// as a failure regardless.
	#[tokio::test]
	async fn axum_ws_rejects_unsupported_subprotocols() {
		let (addr, _rx) = spawn_test_server().await;

		let status = handshake_status(addr, Some("bogus-99, soap")).await;
		assert!(
			status.starts_with("HTTP/1.1 400"),
			"unsupported subprotocols must fail the handshake, got {status:?}",
		);

		// A supported identifier still upgrades.
		let supported = supported_subprotocols();
		let known = supported.first().expect("no supported subprotocols");
		let status = handshake_status(addr, Some(&format!("bogus-99, {known}"))).await;
		assert!(
			status.starts_with("HTTP/1.1 101"),
			"a supported subprotocol must upgrade, got {status:?}",
		);

		// No header at all is the legacy route: upgrade, negotiate via SETUP.
		let status = handshake_status(addr, None).await;
		assert!(
			status.starts_with("HTTP/1.1 101"),
			"offering no subprotocol must still upgrade, got {status:?}",
		);
	}

	/// What a single accepted WebSocket connection negotiated, as seen by the server.
	#[derive(Debug)]
	struct Observed {
		/// Raw `Sec-WebSocket-Protocol` axum selected (keeps the `qmux-XX.` prefix).
		wire: Option<String>,
		/// The moq app ALPN qmux derived from it (prefix stripped, `None` for bare).
		app: Option<String>,
	}

	/// Spawn an axum server that mirrors `serve_ws`'s subprotocol wiring.
	///
	/// Returns its address and a receiver yielding one [`Observed`] per accepted
	/// connection, in acceptance order.
	async fn spawn_test_server() -> (std::net::SocketAddr, mpsc::UnboundedReceiver<Observed>) {
		let (tx, rx) = mpsc::unbounded_channel::<Observed>();

		let route = any(move |ws: WebSocketUpgrade| {
			let tx = tx.clone();
			async move {
				let ws = negotiate_subprotocol(ws)?;
				Ok::<_, StatusCode>(ws.on_upgrade(move |socket| async move {
					let wire = socket.protocol().and_then(|h| h.to_str().ok()).map(str::to_owned);
					let socket = WebSocketAdapter::new(socket);

					let upgraded = qmux::ws::Upgraded::new(socket);
					let upgraded = match wire.as_deref() {
						Some(alpn) => upgraded.with_alpn(alpn),
						None => upgraded,
					};
					let session = upgraded.accept();
					let _ = tx.send(Observed {
						wire,
						app: session.protocol().map(str::to_owned),
					});
					// Hold the session open so the client stays alive long enough
					// to observe the negotiated subprotocol.
					let _ = session.closed().await;
				}))
			}
		});

		let app = Router::new().route("/", route);
		let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
			.await
			.expect("bind listener");
		let addr = listener.local_addr().expect("local addr");
		tokio::spawn(async move {
			axum::serve(listener, app).await.expect("axum serve");
		});
		(addr, rx)
	}

	async fn next_observed(rx: &mut mpsc::UnboundedReceiver<Observed>) -> Observed {
		tokio::time::timeout(Duration::from_secs(5), rx.recv())
			.await
			.expect("server did not report a connection")
			.expect("server channel closed")
	}

	/// Split an advertised `{qmux-XX.}{moq-alpn}` pair into its parts, or `None`
	/// for a bare fallback (`qmux-01`, `qmux-00`, `webtransport`).
	fn split_pair(entry: &str) -> Option<(qmux::Version, &str)> {
		QMUX_VERSIONS
			.iter()
			.find_map(|&v| entry.strip_prefix(v.prefix()).map(|app| (v, app)))
	}

	/// End-to-end regression: a qmux client offering the full moq ALPN list must
	/// land on the newest moq ALPN (`moq_net::ALPNS[0]`) on both sides. A bug in
	/// `supported_subprotocols` or the `with_alpn` plumbing collapses this to
	/// `None` / bare `webtransport`, and moq-net then downgrades to Lite02.
	#[tokio::test]
	async fn axum_ws_negotiates_newest_moq_alpn() {
		let (addr, mut rx) = spawn_test_server().await;

		let session = qmux::Client::new()
			.with_protocols(moq_net::ALPNS.iter().map(|&a| (a, &[] as &[qmux::Version])))
			.connect(&format!("ws://{addr}/"))
			.await
			.expect("qmux client connect");

		assert_eq!(
			session.protocol(),
			Some(newest_moq_alpn()),
			"client side should see the newest moq ALPN, got {:?}",
			session.protocol(),
		);

		let observed = next_observed(&mut rx).await;
		assert_eq!(
			observed.app.as_deref(),
			Some(newest_moq_alpn()),
			"server side should see the newest moq ALPN after with_alpn",
		);

		drop(session);
	}

	/// Every versioned `(qmux, moq)` pair we advertise must be acceptable: a
	/// client offering exactly that pair negotiates it end-to-end. Conversely,
	/// the excluded `qmux-00.moqt-18` pair must never be selected, even when a
	/// client explicitly offers it.
	#[tokio::test]
	async fn every_advertised_pair_is_acceptable() {
		let (addr, mut rx) = spawn_test_server().await;
		let url = format!("ws://{addr}/");

		for entry in supported_subprotocols() {
			// Bare fallbacks can't be offered in isolation via the qmux client API;
			// they're covered by `axum_ws_negotiates_newest_moq_alpn`.
			let Some((version, app)) = split_pair(&entry) else {
				continue;
			};

			let session = qmux::Client::new()
				.with_protocol(app, &[version])
				.connect(&url)
				.await
				.unwrap_or_else(|e| panic!("server rejected advertised pair {entry}: {e}"));

			let observed = next_observed(&mut rx).await;
			assert_eq!(
				observed.wire.as_deref(),
				Some(entry.as_str()),
				"offered advertised pair {entry}, but server negotiated {:?}",
				observed.wire,
			);
			drop(session);
		}

		// The illegal pair is advertised by nobody (moqt-18 requires qmux-01). A client
		// offering only `qmux-00.moqt-18` still carries qmux's default bare fallbacks
		// (`webtransport`, etc.), so the server gracefully downgrades to a bare ALPN
		// rather than failing: we always accept a bare web-transport connection, and
		// there's no other qmux version worth negotiating for moqt-18. It just never
		// lands on `moqt-18`.
		let session = qmux::Client::new()
			.with_protocol("moqt-18", &[qmux::Version::QMux00])
			.connect(&url)
			.await
			.expect("qmux-00.moqt-18 should downgrade to a bare fallback, not fail");

		let observed = next_observed(&mut rx).await;
		assert_eq!(
			observed.app, None,
			"qmux-00.moqt-18 must never be selected; expected a bare downgrade, got {:?}",
			observed.app,
		);
		assert!(
			observed.wire.as_deref().is_none_or(|wire| split_pair(wire).is_none()),
			"the illegal pair must downgrade to a bare fallback, got {:?}",
			observed.wire,
		);
		drop(session);
	}

	/// One leg of an in-memory WebSocket pair.
	///
	/// Reading can be frozen: once the flag is set the stream parks forever
	/// instead of yielding `None`. That is the failure this is here to model --
	/// a peer whose host or network vanished sends no close frame and no FIN, so
	/// the socket stays readable-but-silent indefinitely.
	struct Pipe {
		incoming: mpsc::UnboundedReceiver<tungstenite::Message>,
		outgoing: mpsc::UnboundedSender<tungstenite::Message>,
		frozen: Arc<AtomicBool>,
	}

	impl Pipe {
		fn new(
			incoming: mpsc::UnboundedReceiver<tungstenite::Message>,
			outgoing: mpsc::UnboundedSender<tungstenite::Message>,
			frozen: Arc<AtomicBool>,
		) -> Self {
			Self {
				incoming,
				outgoing,
				frozen,
			}
		}
	}

	impl Stream for Pipe {
		type Item = Result<tungstenite::Message, tungstenite::Error>;

		fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
			if self.frozen.load(Ordering::Relaxed) {
				return Poll::Pending;
			}
			self.incoming.poll_recv(cx).map(|msg| msg.map(Ok))
		}
	}

	impl Sink<tungstenite::Message> for Pipe {
		type Error = tungstenite::Error;

		fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
			Poll::Ready(Ok(()))
		}

		fn start_send(self: Pin<&mut Self>, message: tungstenite::Message) -> Result<(), Self::Error> {
			// A receiver that went away is the peer hanging up, not an error we
			// need to surface: the read half reports the close.
			let _ = self.outgoing.send(message);
			Ok(())
		}

		fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
			Poll::Ready(Ok(()))
		}

		fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
			Poll::Ready(Ok(()))
		}
	}

	/// Regression: a WebSocket peer that goes silent without closing must be
	/// reaped by the keep-alive, not held open forever.
	///
	/// Without `with_keep_alive` on the server's upgrade, nothing in the stack
	/// bounds this: qmux's handshake timeout has already passed, WebSocket has no
	/// idle timeout of its own, and the OS won't probe for hours. The session --
	/// and every broadcast announced through it -- would live until the process
	/// restarts. Time is paused, so the 5s ping / 30s deadline elapse instantly.
	#[tokio::test(start_paused = true)]
	async fn keep_alive_reaps_a_silent_peer() {
		let alpn = format!("{}{}", preferred_qmux_prefix(), newest_moq_alpn());
		let (client_to_server, server_incoming) = mpsc::unbounded_channel();
		let (server_to_client, client_incoming) = mpsc::unbounded_channel();
		let frozen = Arc::new(AtomicBool::new(false));

		let server = tokio::spawn(handle_socket(
			0,
			Pipe::new(server_incoming, server_to_client, frozen.clone()),
			Some(alpn.clone()),
			None,
			None,
			Session::default(),
		));

		// A real qmux peer, so the transport handshake completes and its 10s
		// timeout is out of the picture before we go silent. It never speaks moq,
		// so the server is parked awaiting SETUP -- exactly where an idle
		// publisher's session sits between groups.
		let client = qmux::ws::Upgraded::new(Pipe::new(
			client_incoming,
			client_to_server,
			Arc::new(AtomicBool::new(false)),
		))
		.with_alpn(&alpn)
		.connect();

		// Paused time only advances once every task is idle, so this resolves
		// exactly when both ends have settled.
		tokio::time::sleep(Duration::from_secs(1)).await;
		frozen.store(true, Ordering::Relaxed);

		// Generous versus the 30s deadline: this asserts termination, not timing.
		tokio::time::timeout(Duration::from_secs(300), server)
			.await
			.expect("a silent WebSocket peer must be reaped by the keep-alive")
			.expect("server task panicked")
			.expect_err("the session ends on the keep-alive timeout, never cleanly");

		drop(client);
	}
}
