moq_net/

server.rs

//! Accepting a MoQ session, including the paused handshake that inspects the
//! peer's SETUP before granting origins.

use web_transport_trait::{MaybeSend, MaybeSync};

use crate::origin;
use crate::time::{Clock, Instant};
use crate::{
	ALPN_14, ALPN_15, ALPN_16, ALPN_17, ALPN_18, ALPN_19, ALPN_20, ALPN_21, ALPN_22, ALPN_LITE, ALPN_LITE_03,
	ALPN_LITE_04, ALPN_LITE_05, ALPN_LITE_06, ALPN_LITE_07_WIP, Consume, Error, NEGOTIATED, Role, Session,
	SessionError, Version, Versions,
	coding::{Decode, Encode, Stream},
	ietf, lite, setup, stats,
};

/// A MoQ server session builder.
#[derive(Default, Clone)]
pub struct Server {
	publish: Option<origin::Consumer>,
	subscribe: Option<origin::Producer>,
	stats: stats::Session,
	versions: Versions,
}

impl Server {
	/// A server that neither publishes nor subscribes until configured.
	pub fn new() -> Self {
		Default::default()
	}

	/// Publish to the connected client: the session reads from the given origin
	/// (pass an [`origin::Producer`] or [`origin::Consumer`] by reference) and forwards
	/// its announcements. Omit to publish nothing. Pre-scoped via
	/// [`origin::Producer::scope`] for token-gated relays.
	pub fn with_publisher(mut self, publish: impl Consume<origin::Consumer>) -> Self {
		self.publish = Some(publish.consume());
		self
	}

	/// Subscribe to the connected client: the session writes the broadcasts the
	/// client announces into this [`origin::Producer`]. Omit to subscribe to nothing.
	pub fn with_subscriber(mut self, subscribe: origin::Producer) -> Self {
		self.subscribe = Some(subscribe);
		self
	}

	/// Attach a per-connection [`stats::Session`] context. The session's publish
	/// (egress) and subscribe (ingress) origin handles are tagged with it, so all
	/// traffic counters are attributed through the model for this session's lifetime.
	/// Pass [`stats::Session::default`] (a no-op context) to opt out.
	pub fn with_stats(mut self, stats: stats::Session) -> Self {
		self.stats = stats;
		self
	}

	/// Set both publish and subscribe from one shared [`origin::Producer`].
	pub fn with_origin(self, origin: origin::Producer) -> Self {
		self.with_publisher(&origin).with_subscriber(origin)
	}

	/// Restrict which protocol versions to accept, in preference order.
	/// Defaults to every version this crate supports.
	pub fn with_versions(mut self, versions: Versions) -> Self {
		self.versions = versions;
		self
	}

	/// The configured origin pair, each tagged with the stats context so the
	/// model attributes reads (egress) and writes (ingress) for this session.
	/// One shared context across both halves keeps presence and viewer counts
	/// from double-attributing.
	fn stat_tagged_origins(&self) -> (Option<origin::Consumer>, Option<origin::Producer>) {
		let publish = self.publish.clone().map(|origin| origin.with_stats(self.stats.clone()));
		let subscribe = self
			.subscribe
			.clone()
			.map(|origin| origin.with_stats(self.stats.clone()));
		(publish, subscribe)
	}

	/// Start a lite session on an accepted transport: wire the origins, answer
	/// with our SETUP, and return the session and its driver.
	fn start_lite<S>(
		&self,
		runtime: Clock,
		session: S,
		version: lite::Version,
		client_setup: Option<lite::Setup>,
		peer_hop: Option<crate::Hop>,
	) -> Result<(Session, crate::Driver<S>), Error>
	where
		S: crate::transport::poll::Session,
	{
		let (publish, subscribe) = self.stat_tagged_origins();

		// We report what the transport actually measures; a server never
		// advertises a request Path or Role, and only the dialing side prices a
		// link. Versions without a Setup Stream have nothing to advertise.
		let our_setup = if version.has_setup_stream() {
			lite::Setup {
				probe: lite::ProbeLevel::detect(&session),
				path: None,
				role: None,
				cost: None,
				// Filled by `lite::start` from the attached origin handles.
				hop: None,
			}
		} else {
			lite::Setup::default()
		};

		let start = lite::start(lite::Config {
			runtime: runtime.clone(),
			stats: self.stats.clone(),
			session: session.clone(),
			setup_stream: None,
			publish,
			subscribe,
			peer_hop,
			version,
			our_setup,
			peer_setup: client_setup,
		})?;

		Ok(Session::new(
			runtime,
			session,
			version.into(),
			start.recv_bandwidth,
			crate::driver::Protocol::Lite(Box::new(start.driver)),
			start.goaway,
			Some(start.estimate),
		))
	}

	/// Perform the MoQ handshake for moq-lite only, over any transport.
	///
	/// Same trade as [`Client::connect_lite`](crate::Client::connect_lite): no
	/// thread-affinity bound on the transport, so a pinned `!Send` transport
	/// works, and only a moq-lite ALPN is accepted (anything else is refused
	/// with [`Error::Version`]). Completes the handshake immediately; a caller
	/// gating on the advertised path uses
	/// [`accept_request_lite`](Self::accept_request_lite) instead.
	pub async fn accept_lite<S>(&self, now: Instant, session: S) -> Result<(Session, crate::Driver<S>), Error>
	where
		S: crate::transport::poll::Session,
	{
		self.accept_request_lite(now, session).await?.ok().await
	}

	/// Begin the moq-lite handshake, pausing like
	/// [`accept_request`](Self::accept_request) but for moq-lite ALPNs only,
	/// which is what drops the thread-affinity bounds: a pinned `!Send`
	/// transport can gate on the advertised path too. Anything but a moq-lite
	/// ALPN is refused with [`Error::Version`].
	pub async fn accept_request_lite<S>(&self, now: Instant, session: S) -> Result<Handshake<S>, Error>
	where
		S: crate::transport::poll::Session,
	{
		let mut refused = session.clone();
		self.handshake_lite(now, session)
			.await
			.inspect_err(|err| close(&mut refused, err))
	}

	async fn handshake_lite<S>(&self, now: Instant, mut session: S) -> Result<Handshake<S>, Error>
	where
		S: crate::transport::poll::Session,
	{
		let runtime = Clock::new(now);
		let (path, role, origin, handshake) = match session.protocol() {
			Some(alpn @ (ALPN_LITE_05 | ALPN_LITE_06 | ALPN_LITE_07_WIP)) => {
				let version = match alpn {
					ALPN_LITE_07_WIP => lite::Version::Lite07,
					ALPN_LITE_06 => lite::Version::Lite06,
					_ => lite::Version::Lite05,
				};
				self.versions.select(Version::Lite(version)).ok_or(Error::Version)?;
				// Gate on the client's SETUP: read it before serving so the
				// caller can scope by the advertised path. Seeded back into
				// `start` on `ok()` so PROBE gating resolves without
				// re-reading the (consumed) Setup Stream.
				let (client_setup, client_setup_size) = lite::accept_setup(&mut session, version).await?;
				(
					client_setup.path.clone(),
					client_setup.role,
					client_setup.hop,
					PausedHandshake::LiteSetup {
						session,
						version,
						client_setup,
						client_setup_size,
					},
				)
			}
			Some(ALPN_LITE_04) => {
				self.versions
					.select(Version::Lite(lite::Version::Lite04))
					.ok_or(Error::Version)?;
				(
					None,
					None,
					None,
					PausedHandshake::LiteBare {
						session,
						version: lite::Version::Lite04,
					},
				)
			}
			Some(ALPN_LITE_03) => {
				self.versions
					.select(Version::Lite(lite::Version::Lite03))
					.ok_or(Error::Version)?;
				(
					None,
					None,
					None,
					PausedHandshake::LiteBare {
						session,
						version: lite::Version::Lite03,
					},
				)
			}
			_ => return Err(Error::Version),
		};

		Ok(Handshake {
			path,
			role,
			origin,
			// moq-lite carries no SETUP token.
			token: None,
			assigned_hop: crate::Hop::random(),
			inner: Some(RequestInner {
				server: self.clone(),
				runtime,
				handshake,
			}),
		})
	}

	/// Perform the MoQ handshake as a server, returning the [`Session`] and its [`Driver`](crate::Driver).
	///
	/// Poll the returned driver with nondecreasing time, starting at `now`.
	///
	/// Convenience wrapper over [`accept_request`](Self::accept_request) that
	/// completes the handshake immediately. Use `accept_request` when you need to
	/// inspect the client's advertised path before deciding what to serve.
	pub async fn accept<S>(&self, now: Instant, session: S) -> Result<(Session, crate::Driver<S>), Error>
	where
		S: crate::transport::poll::Boxable,
		S::SendStream: MaybeSync,
		S::RecvStream: MaybeSync,
	{
		self.accept_request(now, session).await?.ok().await
	}

	/// Begin the MoQ handshake, pausing once the client's request path is known so
	/// the caller can authorize/scope before serving.
	///
	/// Reads the client's SETUP (the in-band path lives there on URL-less transports),
	/// then returns a [`Handshake`]: inspect [`path`](Handshake::path), set the origins to
	/// serve, and call [`ok`](Handshake::ok) or [`close`](Handshake::close). Session start
	/// is deferred to `ok()`, so origins set on the handshake always take effect.
	///
	/// The path is surfaced for moq-lite-05 and newer, and every moq-transport
	/// draft we speak; it's empty on versions with no in-band request path (lite 01-04).
	///
	/// A SETUP that fails to parse or negotiate closes the session with the matching code,
	/// so the peer learns why instead of seeing a bare disconnect.
	pub async fn accept_request<S>(&self, now: Instant, session: S) -> Result<Handshake<S>, Error>
	where
		S: crate::transport::poll::Boxable,
		S::SendStream: MaybeSync,
		S::RecvStream: MaybeSync,
	{
		let mut refused = session.clone();
		self.handshake(now, session)
			.await
			.inspect_err(|err| close(&mut refused, err))
	}

	async fn handshake<S>(&self, now: Instant, mut session: S) -> Result<Handshake<S>, Error>
	where
		S: crate::transport::poll::Boxable,
		S::SendStream: MaybeSync,
		S::RecvStream: MaybeSync,
	{
		let runtime = Clock::new(now);
		let (encoding, supported) = match session.protocol() {
			Some(alpn @ (ALPN_22 | ALPN_21 | ALPN_20 | ALPN_19 | ALPN_18 | ALPN_17)) => {
				let draft = match alpn {
					ALPN_22 => ietf::Version::Draft22,
					ALPN_21 => ietf::Version::Draft21,
					ALPN_20 => ietf::Version::Draft20,
					ALPN_19 => ietf::Version::Draft19,
					ALPN_18 => ietf::Version::Draft18,
					_ => ietf::Version::Draft17,
				};

				self.versions.select(Version::Ietf(draft)).ok_or(Error::Version)?;
				return self.accept_ietf_modern(runtime, session, draft).await;
			}
			Some(ALPN_16) => {
				let v = self
					.versions
					.select(Version::Ietf(ietf::Version::Draft16))
					.ok_or(Error::Version)?;
				(v, v.into())
			}
			Some(ALPN_15) => {
				let v = self
					.versions
					.select(Version::Ietf(ietf::Version::Draft15))
					.ok_or(Error::Version)?;
				(v, v.into())
			}
			Some(ALPN_14) => {
				let v = self
					.versions
					.select(Version::Ietf(ietf::Version::Draft14))
					.ok_or(Error::Version)?;
				(v, v.into())
			}
			// Every lite ALPN goes through the same entry point, which is also
			// what a `!Send` transport calls directly.
			Some(ALPN_LITE_07_WIP | ALPN_LITE_06 | ALPN_LITE_05 | ALPN_LITE_04 | ALPN_LITE_03) => {
				return self.handshake_lite(now, session).await;
			}
			Some(ALPN_LITE) | None => {
				let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?;
				(Version::Ietf(ietf::Version::Draft14), supported)
			}
			Some(p) => return Err(Error::UnknownAlpn(p.to_string())),
		};

		// Legacy bidi SETUP exchange (IETF 14-16, lite 01/02). Read the client's
		// SETUP to choose the version; `ok()` sends the server SETUP and starts.
		// Stats attach in `ok()`: a caller may still replace them on the handshake.
		let mut stream = Stream::accept(&mut session, encoding).await?;
		let (mut client, client_setup_size) = stream.reader.decode_sized::<setup::Client>().await?;

		let version = client
			.versions
			.iter()
			.flat_map(|v| Version::try_from(*v).ok())
			.find(|v| supported.contains(v))
			.ok_or(Error::Version)?;

		// Pull the request path and max request ID out now (IETF only) so `ok()`
		// doesn't re-decode the consumed parameters. moq-transport carries the path
		// in its SETUP just like lite-05.
		let (path, token, request_id_max, peer_declared) = match version {
			Version::Ietf(v) => {
				let params = ietf::Parameters::decode(&mut client.parameters, v)?;
				let path = match params.get_bytes(ietf::ParameterBytes::Path) {
					Some(bytes) => Some(
						std::str::from_utf8(bytes)
							.map_err(|_| Error::Decode(crate::DecodeError::InvalidValue))?
							.to_owned(),
					),
					None => None,
				};
				let token = ietf::token::from_setup(&params, v)?;
				let request_id_max = params
					.get_varint(ietf::ParameterVarInt::MaxRequestId)
					.map(ietf::RequestId);
				let peer_declared = ietf::peer::Peer {
					solicit: ietf::solicit::from_setup(&params, v)?,
					hidden: ietf::hidden::from_setup(&params, v),
					..Default::default()
				};
				(path, token, request_id_max, peer_declared)
			}
			Version::Lite(_) => (None, None, None, ietf::peer::Peer::default()),
		};

		Ok(Handshake {
			path,
			role: None,
			origin: None,
			token,
			assigned_hop: crate::Hop::random(),
			inner: Some(RequestInner {
				server: self.clone(),
				runtime,
				handshake: PausedHandshake::Boxed(Box::new(PausedLegacy {
					session,
					stream,
					version,
					request_id_max,
					peer_declared,
					client_setup_size,
				})),
			}),
		})
	}

	/// Read a draft-17/18 client's SETUP (with its request path) off its uni stream,
	/// then pause. `ok()` starts the session and hands the stream back for GOAWAY.
	async fn accept_ietf_modern<S>(
		&self,
		runtime: Clock,
		mut session: S,
		version: ietf::Version,
	) -> Result<Handshake<S>, Error>
	where
		S: crate::transport::poll::Boxable,
		S::SendStream: MaybeSync,
		S::RecvStream: MaybeSync,
	{
		let peer_setup = ietf::accept_setup(&mut session, version).await?;
		Ok(Handshake {
			path: peer_setup.path.clone(),
			role: None,
			// A moq-transport peer only has an identity if it negotiated the MoQ
			// Cluster extension and declared a non-zero Hop ID.
			origin: peer_setup.declared.cluster.hop.filter(|h| *h != crate::Hop::UNKNOWN),
			token: peer_setup.token.clone(),
			assigned_hop: crate::Hop::random(),
			inner: Some(RequestInner {
				server: self.clone(),
				runtime,
				handshake: PausedHandshake::Boxed(Box::new(PausedIetfModern {
					session,
					version,
					peer_setup,
				})),
			}),
		})
	}
}

/// A paused server-side handshake.
///
/// Returned by [`Server::accept_request`] once the peer's advertised
/// [`path`](Self::path) is known but before the session is granted anything. Set
/// the origins to serve, then call [`ok`](Self::ok) to complete the handshake, or
/// [`close`](Self::close) to reject it. Modeled on the WebTransport `Request` in
/// moq-tokio.
pub struct Handshake<S: crate::transport::poll::Session> {
	path: Option<String>,
	role: Option<Role>,
	origin: Option<crate::Hop>,
	token: Option<setup::Token>,
	/// The identity this session's routes are stamped with when the peer declares none
	/// on the wire. Fresh per request unless the caller overrides it
	/// ([`Handshake::with_peer_hop`]).
	assigned_hop: crate::Hop,
	// Taken by `ok`/`close`; `Drop` rejects the handshake if neither ran.
	inner: Option<RequestInner<S>>,
}

/// The parts of a [`Handshake`] consumed by [`Handshake::ok`] / [`Handshake::close`].
struct RequestInner<S: crate::transport::poll::Session> {
	server: Server,
	/// Supplies the clock and timers for the accepted session.
	runtime: Clock,
	handshake: PausedHandshake<S>,
}

/// The handshake state captured at the pause point. Every variant defers its
/// session start to [`Handshake::ok`] so origins set on the handshake still apply.
enum PausedHandshake<S: crate::transport::poll::Session> {
	/// moq-lite 03/04: no Setup Stream.
	LiteBare { session: S, version: lite::Version },
	/// moq-lite 05+: the client's Setup Stream has been read. `ok()` starts the
	/// session, seeding the SETUP back so PROBE gating resolves.
	LiteSetup {
		session: S,
		version: lite::Version,
		client_setup: lite::Setup,
		/// The client SETUP's encoded size, counted under the final stats.
		client_setup_size: usize,
	},
	/// An IETF (or legacy bidi-SETUP) handshake, boxed where its
	/// thread-affinity bounds held. The boxing is what keeps [`Handshake`] and
	/// its lite path free of those bounds: the ietf machinery erases its
	/// futures, which forces a per-target `Send` choice a pinned `!Send`
	/// transport cannot satisfy, so the choice is made here, at construction,
	/// where the caller proved the bounds.
	Boxed(Box<dyn Paused<S>>),
}

type Accept<S> = crate::util::MaybeSendBox<'static, Result<(Session, crate::Driver<S>), Error>>;

/// A paused non-lite handshake. See [`PausedHandshake::Boxed`] for why this is a
/// trait object.
///
/// `MaybeSync` is not decoration: a caller holding a [`Handshake`] across an
/// await behind `&self` (moq-relay authenticates that way) needs
/// `&Handshake: Send`, which is `Handshake: Sync`, which is this.
trait Paused<S: crate::transport::poll::Session>: MaybeSend + MaybeSync {
	/// Complete the handshake with the final server config.
	fn ok(self: Box<Self>, server: Server, runtime: Clock, peer_hop: Option<crate::Hop>) -> Accept<S>;

	/// Reject the handshake, closing the transport with `err`'s wire code.
	fn close(self: Box<Self>, err: Error);
}

/// Modern IETF (17/18): the client's SETUP (with its request path) has been
/// read off its uni stream; `ok` starts the session, handing that stream back
/// for GOAWAY monitoring.
struct PausedIetfModern<S: crate::transport::poll::Session> {
	session: S,
	version: ietf::Version,
	peer_setup: ietf::PeerSetup<S>,
}

impl<S> Paused<S> for PausedIetfModern<S>
where
	S: crate::transport::poll::Boxable,
	S::SendStream: MaybeSync,
	S::RecvStream: MaybeSync,
{
	fn ok(self: Box<Self>, server: Server, runtime: Clock, peer_hop: Option<crate::Hop>) -> Accept<S> {
		use crate::util::MaybeBoxedExt as _;
		async move {
			let Self {
				session,
				version,
				peer_setup,
			} = *self;
			server.stats.control(stats::Control::Session, peer_setup.size, true);
			let (publish, subscribe) = server.stat_tagged_origins();

			// The client's SETUP was read at the pause; hand the stream back
			// for GOAWAY. A server never advertises a path, hence `None`.
			let (protocol, goaway) = ietf::start(ietf::Config {
				runtime: runtime.clone(),
				stats: server.stats.clone(),
				session: session.clone(),
				setup: None,
				request_id_max: None,
				client: false,
				publish,
				subscribe,
				peer_hop,
				// Only the dialing side prices a link.
				cost: None,
				version,
				path: None,
				authority: None,
				peer_setup_stream: Some(peer_setup.stream),
				peer_declared: Some(peer_setup.declared),
				early_unis: peer_setup.early,
			})?;
			tracing::debug!(?version, "connected");
			Ok(Session::new(
				runtime,
				session,
				version.into(),
				None,
				crate::driver::Protocol::Ietf(protocol),
				goaway,
				None,
			))
		}
		.maybe_boxed()
	}

	fn close(mut self: Box<Self>, err: Error) {
		close(&mut self.session, &err);
	}
}

/// Legacy IETF (draft 14-16) and lite 01/02: the client SETUP has been read
/// off the bidi stream (including its request path) but the server SETUP
/// hasn't been sent; `ok` finishes it.
struct PausedLegacy<S: crate::transport::poll::Session> {
	session: S,
	stream: Stream<S, Version>,
	version: Version,
	request_id_max: Option<ietf::RequestId>,
	/// What the client's SETUP declared, for the options `ok` acts on.
	peer_declared: ietf::peer::Peer,
	/// The client SETUP's encoded size, counted under the final stats.
	client_setup_size: usize,
}

impl<S> Paused<S> for PausedLegacy<S>
where
	S: crate::transport::poll::Boxable,
	S::SendStream: MaybeSync,
	S::RecvStream: MaybeSync,
{
	fn ok(self: Box<Self>, server: Server, runtime: Clock, peer_hop: Option<crate::Hop>) -> Accept<S> {
		use crate::util::MaybeBoxedExt as _;
		async move {
			let Self {
				session,
				stream,
				version,
				request_id_max,
				peer_declared,
				client_setup_size,
			} = *self;
			let mut stream = stream.with_stats(&server.stats);
			server.stats.control(stats::Control::Session, client_setup_size, true);
			let (publish, subscribe) = server.stat_tagged_origins();

			// Encode parameters using the version-appropriate type.
			let parameters = match version {
				Version::Ietf(v) => {
					let mut parameters = ietf::Parameters::default();
					parameters.set_varint(ietf::ParameterVarInt::MaxRequestId, u32::MAX as u64);
					parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec());
					ietf::solicit::into_setup(&mut parameters, v);
					ietf::hidden::into_setup(&mut parameters, v);
					parameters.encode_bytes(v)?
				}
				Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?,
			};

			let server_setup = setup::Server {
				version: version.into(),
				parameters,
			};
			stream.writer.encode(&server_setup).await?;

			let (recv_bw, protocol, goaway, estimate) = match version {
				Version::Lite(v) => {
					let stream = stream.with_version(v);
					// Pre-lite-05: no Setup Stream, so nothing to advertise or seed.
					let start = lite::start(lite::Config {
						runtime: runtime.clone(),
						stats: server.stats.clone(),
						session: session.clone(),
						setup_stream: Some(stream),
						publish,
						subscribe,
						peer_hop,
						version: v,
						our_setup: lite::Setup::default(),
						peer_setup: None,
					})?;
					(
						start.recv_bandwidth,
						crate::driver::Protocol::Lite(Box::new(start.driver)),
						start.goaway,
						Some(start.estimate),
					)
				}
				Version::Ietf(v) => {
					let stream = stream.with_version(v);
					// Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back.
					let (protocol, goaway) = ietf::start(ietf::Config {
						runtime: runtime.clone(),
						stats: server.stats.clone(),
						session: session.clone(),
						setup: Some(stream),
						request_id_max,
						client: false,
						publish,
						subscribe,
						peer_hop,
						cost: None,
						version: v,
						path: None,
						authority: None,
						peer_setup_stream: None,
						peer_declared: Some(peer_declared),
						early_unis: Vec::new(),
					})?;
					(None, crate::driver::Protocol::Ietf(protocol), goaway, None)
				}
			};

			Ok(Session::new(
				runtime, session, version, recv_bw, protocol, goaway, estimate,
			))
		}
		.maybe_boxed()
	}

	fn close(mut self: Box<Self>, err: Error) {
		close(&mut self.session, &err);
	}
}

impl<S> Handshake<S>
where
	S: crate::transport::poll::Session,
{
	/// The request path the client advertised in its SETUP.
	///
	/// Empty when the client advertised none: either it sent an empty path, or the
	/// version carries none in-band (lite 01-04). Those mean the same thing, so the
	/// wire distinction isn't surfaced. Populated for moq-lite-05 and newer,
	/// and every moq-transport draft we speak. See the note on [`Server::accept_request`].
	pub fn path(&self) -> &str {
		self.path.as_deref().unwrap_or("")
	}

	/// The single [`Role`] the client advertised in its SETUP, or `None` for a
	/// bidirectional session.
	///
	/// Only moq-lite-05 and newer carry a role, so `None` covers three cases
	/// that the wire doesn't distinguish: an older version, a client that omitted the parameter, and a
	/// client that explicitly advertised both directions. All three mean the same thing
	/// (the client may publish and subscribe), so authorize on what the token grants.
	/// See the note on [`Server::accept_request`].
	pub fn role(&self) -> Option<Role> {
		self.role
	}

	/// The Hop ID declared by the peer, when the negotiated protocol carries one.
	///
	/// A moq-lite-05+ endpoint declares this when it attaches a publish or subscribe
	/// origin; a `moqt-17`+ endpoint declares it via the MoQ Cluster extension. Older
	/// versions and endpoints without one return `None`.
	///
	/// Self-declared, so treat it as a correlation hint rather than an
	/// authenticated identity: authorize on the token or client certificate.
	pub fn peer_hop(&self) -> Option<crate::Hop> {
		self.origin
	}

	/// The credential the client presented in its SETUP's `AUTHORIZATION TOKEN` option.
	///
	/// Only moq-transport carries one, so moq-lite sessions return `None`. The transport
	/// has not verified it: authorize on it the way you would a URL token.
	pub fn token(&self) -> Option<&setup::Token> {
		self.token.as_ref()
	}

	/// Publish to the connected client. Overrides any value from the [`Server`]
	/// builder; typically set after inspecting [`path`](Self::path).
	pub fn with_publisher(mut self, publish: impl Consume<origin::Consumer>) -> Self {
		self.inner_mut().server.publish = Some(publish.consume());
		self
	}

	/// Subscribe to the connected client. Overrides any value from the [`Server`] builder.
	pub fn with_subscriber(mut self, subscribe: origin::Producer) -> Self {
		self.inner_mut().server.subscribe = Some(subscribe);
		self
	}

	/// Assign the identity this peer's routes are attributed to, overriding the fresh
	/// per-session default.
	///
	/// Only for a peer whose identity the server has actually established, such as one
	/// authenticated by mTLS or a token ([`crate::Client::with_peer_hop`] is the
	/// dialing-side equivalent). An identity the peer declares on the wire still wins.
	///
	/// Two sessions given the same origin are treated as one endpoint: routes learned
	/// from either are kept off both, and content arriving on either is interchangeable
	/// with the other's. That is the point when they really are one peer reconnecting or
	/// running redundant links, and a bug otherwise. Derive it from the authenticated
	/// identity, never from something coarser like the remote address.
	pub fn with_peer_hop(mut self, hop: crate::Hop) -> Self {
		self.assigned_hop = hop;
		self
	}

	/// Set the per-connection [`stats::Session`] context. Overrides any value from the
	/// [`Server`] builder.
	pub fn with_stats(mut self, stats: stats::Session) -> Self {
		self.inner_mut().server.stats = stats;
		self
	}

	fn inner_mut(&mut self) -> &mut RequestInner<S> {
		self.inner.as_mut().expect("request already responded")
	}

	/// Accept the session, returning the [`Session`] and its [`Driver`](crate::Driver).
	///
	/// Poll or spawn the returned driver to run the session.
	pub async fn ok(mut self) -> Result<(Session, crate::Driver<S>), Error> {
		let peer_hop = Some(self.assigned_hop);
		let RequestInner {
			server,
			runtime,
			handshake,
		} = self.inner.take().expect("request already responded");

		match handshake {
			PausedHandshake::LiteBare { session, version } => {
				server.start_lite(runtime, session, version, None, peer_hop)
			}
			PausedHandshake::LiteSetup {
				session,
				version,
				client_setup,
				client_setup_size,
			} => {
				server.stats.control(stats::Control::Session, client_setup_size, true);
				server.start_lite(runtime, session, version, Some(client_setup), peer_hop)
			}
			PausedHandshake::Boxed(paused) => paused.ok(server, runtime, peer_hop).await,
		}
	}

	/// Reject the session, closing the transport with `err`'s wire code.
	pub fn close(mut self, err: Error) {
		let inner = self.inner.take().expect("request already responded");
		inner.close(err);
	}
}

impl<S: crate::transport::poll::Session> RequestInner<S> {
	fn close(self, err: Error) {
		let mut session = match self.handshake {
			PausedHandshake::LiteBare { session, .. } => session,
			PausedHandshake::LiteSetup { session, .. } => session,
			PausedHandshake::Boxed(paused) => return paused.close(err),
		};
		close(&mut session, &err);
	}
}

/// Close `session` with `err`'s wire code.
fn close<S: crate::transport::poll::Session>(session: &mut S, err: &Error) {
	session.close(SessionError::from(err).to_code(), &err.to_string());
}

impl<S: crate::transport::poll::Session> Drop for Handshake<S> {
	// A dropped request would otherwise leave the client hanging until its idle
	// timeout: it already sent SETUP and is waiting on a response. Reject loudly.
	fn drop(&mut self) {
		if let Some(inner) = self.inner.take() {
			tracing::warn!("Handshake dropped without ok() or close(); rejecting the session");
			inner.close(Error::Cancel);
		}
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use crate::Hop;
	use crate::model::ProduceTest;
	use std::{
		collections::VecDeque,
		sync::{Arc, Mutex},
	};

	use crate::ALPN_LITE_05;
	use bytes::Bytes;

	fn occurrences(log: &crate::lite::test_transport::Log, needle: &[u8]) -> usize {
		let writes = log.writes.lock().unwrap();
		writes.windows(needle.len()).filter(|window| *window == needle).count()
	}

	#[derive(Debug, Clone, Default)]
	struct FakeError;
	impl std::fmt::Display for FakeError {
		fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
			write!(f, "fake transport error")
		}
	}
	impl std::error::Error for FakeError {}
	impl web_transport_trait::Error for FakeError {
		fn session_error(&self) -> Option<(u32, String)> {
			Some((0, "closed".to_string()))
		}
	}

	/// A session that replays a queue of streams (each a `Vec<u8>`) in order from
	/// `accept_uni` and `accept_bi`, and records the code it was closed with; everything
	/// else is inert.
	#[derive(Clone)]
	struct FakeSession {
		protocol: Option<&'static str>,
		uni: Arc<Mutex<VecDeque<Vec<u8>>>>,
		bi: Arc<Mutex<VecDeque<Vec<u8>>>>,
		closed: Arc<Mutex<Option<u32>>>,
	}

	impl FakeSession {
		fn new(protocol: &'static str, uni: impl IntoIterator<Item = Vec<u8>>) -> Self {
			Self {
				protocol: Some(protocol),
				uni: Arc::new(Mutex::new(uni.into_iter().collect())),
				bi: Default::default(),
				closed: Default::default(),
			}
		}

		fn with_bi(self, bi: Vec<u8>) -> Self {
			self.bi.lock().unwrap().push_back(bi);
			self
		}

		fn closed(&self) -> Option<u32> {
			*self.closed.lock().unwrap()
		}
	}

	impl web_transport_trait::poll::Session for FakeSession {
		type SendStream = FakeSend;
		type RecvStream = FakeRecv;
		type Error = FakeError;

		fn poll_accept_uni(
			&mut self,
			_cx: &mut std::task::Context<'_>,
		) -> std::task::Poll<Result<Self::RecvStream, Self::Error>> {
			match self.uni.lock().unwrap().pop_front() {
				Some(data) => std::task::Poll::Ready(Ok(FakeRecv { data: data.into() })),
				None => std::task::Poll::Pending,
			}
		}
		fn poll_accept_bi(
			&mut self,
			_cx: &mut std::task::Context<'_>,
		) -> std::task::Poll<Result<(Self::SendStream, Self::RecvStream), Self::Error>> {
			match self.bi.lock().unwrap().pop_front() {
				Some(data) => std::task::Poll::Ready(Ok((FakeSend, FakeRecv { data: data.into() }))),
				None => std::task::Poll::Pending,
			}
		}
		fn poll_open_bi(
			&mut self,
			_cx: &mut std::task::Context<'_>,
		) -> std::task::Poll<Result<(Self::SendStream, Self::RecvStream), Self::Error>> {
			std::task::Poll::Pending
		}
		fn poll_open_uni(
			&mut self,
			_cx: &mut std::task::Context<'_>,
		) -> std::task::Poll<Result<Self::SendStream, Self::Error>> {
			std::task::Poll::Pending
		}
		fn poll_send_datagram(
			&mut self,
			_cx: &mut std::task::Context<'_>,
			_payload: &[u8],
		) -> std::task::Poll<Result<(), Self::Error>> {
			std::task::Poll::Ready(Ok(()))
		}
		fn poll_recv_datagram(
			&mut self,
			_cx: &mut std::task::Context<'_>,
		) -> std::task::Poll<Result<Bytes, Self::Error>> {
			std::task::Poll::Pending
		}
		fn max_datagram_size(&self) -> usize {
			1200
		}
		fn protocol(&self) -> Option<&str> {
			self.protocol
		}
		fn close(&mut self, code: u32, _reason: &str) {
			self.closed.lock().unwrap().get_or_insert(code);
		}
		fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Error> {
			std::task::Poll::Pending
		}
		fn stats(&self) -> impl web_transport_trait::Stats {
			web_transport_trait::StatsUnavailable
		}
	}

	#[derive(Clone, Default)]
	struct FakeSend;
	impl web_transport_trait::poll::SendStream for FakeSend {
		type Error = FakeError;
		fn poll_write(
			&mut self,
			_cx: &mut std::task::Context<'_>,
			buf: &[u8],
		) -> std::task::Poll<Result<usize, Self::Error>> {
			std::task::Poll::Ready(Ok(buf.len()))
		}
		fn set_priority(&mut self, _order: u8) {}
		fn finish(&mut self) -> Result<(), Self::Error> {
			Ok(())
		}
		fn reset(&mut self, _code: u32) {}
		fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
			std::task::Poll::Ready(Ok(()))
		}
	}

	struct FakeRecv {
		data: VecDeque<u8>,
	}
	impl web_transport_trait::poll::RecvStream for FakeRecv {
		type Error = FakeError;
		fn poll_read(
			&mut self,
			_cx: &mut std::task::Context<'_>,
			dst: &mut [u8],
		) -> std::task::Poll<Result<Option<usize>, Self::Error>> {
			if self.data.is_empty() {
				return std::task::Poll::Ready(Ok(None));
			}
			let size = dst.len().min(self.data.len());
			for slot in dst.iter_mut().take(size) {
				*slot = self.data.pop_front().unwrap();
			}
			std::task::Poll::Ready(Ok(Some(size)))
		}
		fn stop(&mut self, _code: u32) {}
		fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
			std::task::Poll::Ready(Ok(()))
		}
	}

	/// Encode a lite-05 Setup Stream: the `DataType::Setup` tag then the SETUP message.
	fn lite05_setup(path: Option<&str>, role: Option<Role>, hop: Option<Hop>) -> Vec<u8> {
		let v = lite::Version::Lite05;
		let mut buf = Vec::new();
		lite::DataType::Setup.encode(&mut buf, v).unwrap();
		lite::Setup {
			probe: lite::ProbeLevel::None,
			path: path.map(str::to_string),
			role,
			cost: None,
			hop,
		}
		.encode(&mut buf, v)
		.unwrap();
		buf
	}

	/// Encode a draft-17+ Setup Stream: the unified SETUP message, whose parameters
	/// carry the request path the same way lite-05's does.
	fn ietf_setup(version: ietf::Version, path: Option<&str>) -> Vec<u8> {
		let mut params = ietf::Parameters::default();
		if let Some(path) = path {
			params.set_bytes(ietf::ParameterBytes::Path, path.as_bytes().to_vec());
		}
		ietf_setup_with(version, params)
	}

	fn ietf_setup_with(version: ietf::Version, params: ietf::Parameters) -> Vec<u8> {
		let parameters = params.encode_bytes(version).unwrap();

		let mut buf = Vec::new();
		setup::Setup { parameters }
			.encode(&mut buf, crate::Version::Ietf(version))
			.unwrap();
		buf
	}

	/// Encode a draft 14-16 CLIENT_SETUP, sent on the control bidi stream.
	fn legacy_setup(version: ietf::Version, params: ietf::Parameters) -> Vec<u8> {
		let mut buf = Vec::new();
		setup::Client {
			versions: crate::coding::Versions::from([crate::Version::Ietf(version).into()]),
			parameters: params.encode_bytes(version).unwrap(),
		}
		.encode(&mut buf, crate::Version::Ietf(version))
		.unwrap();
		buf
	}

	fn setup_token() -> setup::Token {
		setup::Token {
			kind: setup::Token::OUT_OF_BAND,
			value: vec![0x00, 0xff, b'j', b'w', b't'],
		}
	}

	fn token_params(version: ietf::Version) -> ietf::Parameters {
		let mut params = ietf::Parameters::default();
		ietf::token::into_setup(&mut params, &setup_token(), version).unwrap();
		params
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_exposes_the_setup_token() {
		let modern = FakeSession::new(
			ALPN_19,
			[ietf_setup_with(
				ietf::Version::Draft19,
				token_params(ietf::Version::Draft19),
			)],
		);
		let legacy = FakeSession::new(ALPN_16, []).with_bi(legacy_setup(
			ietf::Version::Draft16,
			token_params(ietf::Version::Draft16),
		));
		for (name, session) in [("draft-19", modern), ("draft-16", legacy)] {
			let request = Server::new()
				.accept_request(tokio::time::Instant::now().into_std(), session)
				.await
				.unwrap();
			assert_eq!(request.token(), Some(&setup_token()), "{name}");
		}
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_without_a_token_reports_none() {
		let ietf = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, None)]);
		let lite = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]);
		for (name, session) in [("draft-19", ietf), ("lite-05", lite)] {
			let request = Server::new()
				.accept_request(tokio::time::Instant::now().into_std(), session)
				.await
				.unwrap();
			assert_eq!(request.token(), None, "{name}");
		}
	}

	/// A SETUP the server refuses closes the session with the code naming why, on both
	/// the draft-17+ uni stream and the draft 14-16 bidi stream.
	#[tokio::test(start_paused = true)]
	async fn a_refused_setup_token_closes_with_its_code() {
		let delete = [0x0, 0x7]; // DELETE alias 7
		let truncated = [0x3]; // USE_VALUE with no Token Type
		for (raw, code) in [
			(&delete[..], SessionError::ProtocolViolation),
			(&truncated[..], SessionError::KeyValueFormatting),
		] {
			let mut params = ietf::Parameters::default();
			params.set_bytes(ietf::ParameterBytes::AuthorizationToken, raw.to_vec());

			let modern = FakeSession::new(ALPN_19, [ietf_setup_with(ietf::Version::Draft19, params.clone())]);
			let legacy = FakeSession::new(ALPN_16, []).with_bi(legacy_setup(ietf::Version::Draft16, params));
			for (name, session) in [("draft-19", modern), ("draft-16", legacy)] {
				let result = Server::new()
					.accept_request(tokio::time::Instant::now().into_std(), session.clone())
					.await;
				assert!(result.is_err(), "{name}");
				assert_eq!(session.closed(), Some(code.to_code()), "{name} {code}");
			}
		}
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_reads_ietf_path() {
		// Every draft-17+ version gates on the SETUP stream before starting, so the
		// path is known at authorization time just like lite-05.
		for (alpn, version) in [
			(ALPN_17, ietf::Version::Draft17),
			(ALPN_18, ietf::Version::Draft18),
			(ALPN_19, ietf::Version::Draft19),
		] {
			let session = FakeSession::new(alpn, [ietf_setup(version, Some("/team/room"))]);
			let request = Server::new()
				.accept_request(tokio::time::Instant::now().into_std(), session)
				.await
				.unwrap();
			assert_eq!(request.path(), "/team/room", "{alpn}");
		}
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_ietf_without_path_is_empty() {
		let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, None)]);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.path(), "");
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_ietf_empty_path_is_accepted() {
		let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, Some(""))]);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.path(), "");
	}

	/// Encode a lite-05 GROUP uni stream header (just the `DataType::Group` tag).
	fn lite05_group() -> Vec<u8> {
		let mut buf = Vec::new();
		lite::DataType::Group.encode(&mut buf, lite::Version::Lite05).unwrap();
		buf
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_reads_lite05_path() {
		let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/team/room"), None, None)]);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.path(), "/team/room");
		assert_eq!(request.role(), None, "a client that omits the role is bidirectional");
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_lite05_without_path_is_empty() {
		let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.path(), "");
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_lite05_empty_path_is_accepted() {
		// An empty path is valid on the wire and means the same as omitting it, so a
		// client that wants the root doesn't have to special-case the parameter.
		let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some(""), None, None)]);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.path(), "");
	}

	#[tokio::test(start_paused = true)]
	async fn repeated_lite_setup_parameter_closes_the_session() {
		// SETUP type, five-byte message, then two empty parameters with id 0.
		let session = FakeSession::new(ALPN_LITE_05, [vec![1, 5, 2, 0, 0, 0, 0]]);
		let result = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session.clone())
			.await;
		assert!(result.is_err());
		assert_eq!(session.closed(), Some(SessionError::ProtocolViolation.to_code()));
	}

	#[tokio::test(start_paused = true)]
	async fn accepted_lite_setup_refuses_a_second_setup_stream() {
		let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]);
		let transport = session.clone();
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		let (_session, mut driver) = request.ok().await.unwrap();
		transport.uni.lock().unwrap().push_back(vec![1]);
		let _ = driver.poll(tokio::time::Instant::now().into_std(), &kio::Waiter::noop());
		assert_eq!(transport.closed(), Some(SessionError::ProtocolViolation.to_code()));
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_reads_lite05_role() {
		let session = FakeSession::new(
			ALPN_LITE_05,
			[lite05_setup(Some("/team/room"), Some(Role::Publisher), None)],
		);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.role(), Some(Role::Publisher));
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_skips_uni_stream_before_setup() {
		// A GROUP racing ahead of the SETUP is STOP_SENDING-ed and skipped; the gate
		// keeps reading until it finds the SETUP.
		let session = FakeSession::new(
			ALPN_LITE_05,
			[lite05_group(), lite05_setup(Some("/team/room"), None, None)],
		);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.path(), "/team/room");
	}

	#[tokio::test(start_paused = true)]
	async fn accept_request_reads_lite05_peer_hop() {
		let hop = Hop::new(42).unwrap();
		let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, Some(hop))]);
		let request = Server::new()
			.accept_request(tokio::time::Instant::now().into_std(), session)
			.await
			.unwrap();
		assert_eq!(request.peer_hop(), Some(hop));
	}

	#[tokio::test(start_paused = true)]
	async fn anonymous_peer_hop_filters_routes_from_server_session() {
		let other = Hop::new(778).unwrap();
		let origin = crate::origin::Config::new(Hop::new(1).unwrap()).produce();

		let gate = kio::Producer::new(true);
		let transport = crate::lite::test_transport::SinkSession::gated_bi(gate.consume());
		let log = transport.log.clone();
		let version = ietf::Version::Draft18;
		let request = Handshake {
			path: None,
			role: None,
			origin: None,
			token: None,
			assigned_hop: Hop::random(),
			inner: Some(RequestInner {
				server: Server::new().with_publisher(&origin),
				runtime: Clock::new(tokio::time::Instant::now().into_std()),
				handshake: PausedHandshake::Boxed(Box::new(PausedIetfModern {
					session: transport,
					version,
					peer_setup: ietf::PeerSetup {
						stream: crate::coding::Reader::new(
							crate::lite::test_transport::PendingRecv,
							Version::Ietf(version),
						),
						size: 0,
						path: None,
						token: None,
						declared: ietf::peer::Peer::default(),
						early: Vec::new(),
					},
				})),
			}),
		};
		let assigned = request.assigned_hop;

		let mut echoed_hops = crate::Hops::new();
		echoed_hops.push(crate::Hop::UNKNOWN).unwrap();
		let _echoed = origin
			.announce(
				"echoed-route",
				crate::origin::Route::default()
					.with_hops(echoed_hops)
					.with_via(assigned),
			)
			.unwrap();

		let mut local_hops = crate::Hops::new();
		local_hops.push(other).unwrap();
		let _local = origin
			.announce("local-route", crate::origin::Route::default().with_hops(local_hops))
			.unwrap();

		let (session, driver) = request.ok().await.unwrap();
		tokio::spawn(crate::time::run(driver));

		for _ in 0..100 {
			if occurrences(&log, b"local-route") > 0 {
				break;
			}
			tokio::time::sleep(std::time::Duration::from_millis(1)).await;
		}

		assert_eq!(occurrences(&log, b"echoed-route"), 0);
		assert_eq!(occurrences(&log, b"local-route"), 1);
		drop(session);
	}
}