web-faith 1.1.0

//! The send path.

use std::{
	sync::{
		Arc,
		atomic::{AtomicBool, Ordering},
	},
	time::Instant,
};

use reqwest::{
	Method, StatusCode,
	header::{CONTENT_ENCODING, CONTENT_TYPE, HeaderName, HeaderValue},
	tls::TlsInfo,
};
use reqwest_middleware::ClientWithMiddleware;

#[cfg(feature = "connection-tracking")]
use hyper_util::client::legacy::connect::HttpInfo;

#[cfg(feature = "cache")]
use http_cache_reqwest::CacheMode;

#[cfg(feature = "encoding")]
use reqwest::header::ACCEPT_ENCODING;

#[cfg(feature = "encoding")]
use web_faith_encoding::{
	Coding, ContentEncoding, request as encoding_request,
	response::{AcceptEncoding, DEFAULT_ACCEPT_ENCODING},
};

use crate::{
	agent::Agent,
	body::{BodyParts, BodyShared},
	error::{FaithError, FaithErrorKind},
	request::{Credentials, NORMALISED_METHODS, PRIORITY, QUERY, RequestBody, RequestOptions},
	response::{PeerInformation, Response, TrailersSlot},
	timing::{RequestTiming, TimingSlot, alpn_protocol_id},
};

/// Send a request on `agent`, and build the response it produces.
///
/// `client` is the handle taken when the request was issued, so one issued just before the agent
/// closes still runs to completion. `abort` cancels the request by resolving first.
// spec:AGENT
pub async fn send(
	agent: &Agent,
	client: ClientWithMiddleware,
	url: &str,
	options: RequestOptions,
	body: RequestBody,
	abort: Option<impl Future<Output = ()>>,
) -> Result<Response, FaithError> {
	let method = options.method.as_deref().unwrap_or("GET");
	// spec:REQ#method-and-headers
	let method = NORMALISED_METHODS
		.into_iter()
		.find(|normalised| normalised.eq_ignore_ascii_case(method))
		.unwrap_or(method);

	let method =
		Method::from_bytes(method.as_bytes()).map_err(|_| FaithErrorKind::InvalidMethod)?;
	let is_head = method == Method::HEAD;
	// Captured before the builder takes the method (spec:REQ#body).
	let is_query = method.as_str() == QUERY;

	let mut parsed_url = reqwest::Url::parse(&url).map_err(|_| FaithErrorKind::InvalidUrl)?;

	// A `compress` naming no coding Faith can compress in is misuse whether or not the
	// request turns out to carry a body, so it is refused before anything else looks at
	// it (spec:ENC#compressing-a-request-body).
	#[cfg(feature = "encoding")]
	let compress = options
		.compress
		.as_deref()
		.map(|value| {
			Coding::from_option(value).ok_or_else(|| {
				FaithError::new(
					FaithErrorKind::InvalidCompression,
					format!(
						"compress: {value:?} names no coding; expected gzip, deflate, br, or zstd"
					),
				)
			})
		})
		.transpose()?;

	// Handle credentials based on credentials option
	if options.credentials == Credentials::Omit {
		// Remove credentials from URL if omit is specified
		let _ = parsed_url.set_username("");
		let _ = parsed_url.set_password(None);
	}

	// The stamp rides along in the request's extensions for the middleware to fill in;
	let mut request = client.request(method, parsed_url.clone());
	#[cfg(feature = "cache")]
	{
		request = request.with_extension(CacheMode::from(options.cache));
	}

	if let Some(headers) = &options.headers {
		for (key, value) in headers {
			// Skip Cookie header if credentials is omit
			if options.credentials == Credentials::Omit && key.eq_ignore_ascii_case("cookie") {
				continue;
			}

			// Validate header name and value before adding to request
			let header_name = HeaderName::from_bytes(key.as_bytes()).map_err(|_| {
				FaithError::new(
					FaithErrorKind::InvalidHeader,
					format!("invalid header name: {key}"),
				)
			})?;
			let header_value = HeaderValue::from_str(value).map_err(|_| {
				FaithError::new(
					FaithErrorKind::InvalidHeader,
					format!("invalid header value: {value}"),
				)
			})?;

			// Faith's coding is layered on top of what the caller declares, and reqwest's
			// builder appends rather than replaces, so passing the caller's value through
			// here would put a second `Content-Encoding` beside the joined one -- the same
			// list read twice over (spec:ENC#what-a-compressed-request-sends). The value is
			// still validated above, then withheld and re-emitted once below.
			#[cfg(feature = "encoding")]
			if compress.is_some() && header_name == CONTENT_ENCODING {
				continue;
			}

			request = request.header(header_name, header_value);
		}
	}

	// A `Content-Type` the request declares, else one the agent declares, else the type the
	// body's kind implies. The derived type sits last because it describes a default the fetch
	// standard extracts rather than anything the caller asked for, so an agent that types every
	// body it sends keeps doing so (spec:REQ#body).
	let declares_content_type = options.headers.as_ref().is_some_and(|headers| {
		headers
			.iter()
			.any(|(name, _)| name.eq_ignore_ascii_case(CONTENT_TYPE.as_str()))
	}) || agent.has_default_content_type;

	if !declares_content_type && let Some(derived) = options.body_content_type.as_deref() {
		let value = HeaderValue::from_str(derived).map_err(|_| {
			FaithError::new(
				FaithErrorKind::InvalidHeader,
				format!("invalid Content-Type derived from the body: {derived}"),
			)
		})?;
		request = request.header(CONTENT_TYPE, value);
	}

	// `QUERY` carries its query content in the body, and content nothing describes cannot be
	// read (spec:REQ#body). Refused here rather than sent for the origin to reject.
	if is_query
		&& !matches!(body, RequestBody::None)
		&& !declares_content_type
		&& options.body_content_type.is_none()
	{
		return Err(FaithError::new(
			FaithErrorKind::MissingContentType,
			"a QUERY request carrying a body must declare a Content-Type describing it",
		));
	}

	// The `Content-Encoding` the caller declared: their own, else the agent's, per-request
	// headers winning per name as they do generally (spec: REQ).
	// Several lines are the one list, so they are joined as they are read.
	#[cfg(feature = "encoding")]
	let declared_content_encoding = compress.as_ref().and_then(|_| {
		let from_request = options.headers.as_ref().and_then(|headers| {
			let declared = headers
				.iter()
				.filter(|(name, _)| name.eq_ignore_ascii_case(CONTENT_ENCODING.as_str()))
				.map(|(_, value)| value.as_str())
				.collect::<Vec<_>>();
			(!declared.is_empty()).then(|| declared.join(", "))
		});
		from_request.or_else(|| {
			agent
				.default_content_encoding
				.as_ref()
				.and_then(|value| value.to_str().ok().map(str::to_owned))
		})
	});

	// The request's `Accept-Encoding` governs which codings Faith decodes on the way
	// back (spec: ENC): a value on the request, else one inherited from the agent's
	// default headers, else the default Faith sends itself. Neither the request nor the
	// agent advertising a value means nothing beneath Faith adds one now that it owns
	// the codings, so Faith sends the default explicitly.
	#[cfg(feature = "encoding")]
	let request_accept_encoding = options.headers.as_ref().and_then(|headers| {
		headers
			.iter()
			.find(|(name, _)| name.eq_ignore_ascii_case("accept-encoding"))
			.map(|(_, value)| value.clone())
	});
	#[cfg(feature = "encoding")]
	let accept_encoding = AcceptEncoding::from(
		&*request_accept_encoding
			.clone()
			.or_else(|| {
				agent
					.default_accept_encoding
					.as_ref()
					.and_then(|value| value.to_str().ok().map(str::to_owned))
			})
			.unwrap_or_else(|| DEFAULT_ACCEPT_ENCODING.to_owned()),
	);
	#[cfg(feature = "encoding")]
	if request_accept_encoding.is_none() && agent.default_accept_encoding.is_none() {
		request = request.header(
			ACCEPT_ENCODING,
			HeaderValue::from_static(DEFAULT_ACCEPT_ENCODING),
		);
	}

	// The `priority` option is a hint, so a `Priority` header the caller wrote, or one
	// among the agent's default headers, wins over the value derived from it
	// (spec: REQ#request-priority). The agent's defaults are consulted here rather than
	// left to reqwest: it fills a default header in only where the request carries none
	// of that name, so setting the derived value would displace the agent's own.
	if let Some(urgency) = options.priority
		&& !agent.has_default_priority
		&& !options.headers.as_ref().is_some_and(|headers| {
			headers
				.iter()
				.any(|(name, _)| name.eq_ignore_ascii_case(PRIORITY))
		}) {
		request = request.header(
			HeaderName::from_static(PRIORITY),
			HeaderValue::from_static(urgency),
		);
	}

	// The coding actually applied, which is `compress` only where there was a body to
	// apply it to: the option does nothing on a request carrying none, so no
	// `Content-Encoding` describes bytes that were never sent
	// (spec:ENC#compressing-a-request-body).
	#[cfg(feature = "encoding")]
	let mut applied_coding = None;

	// Handle body: prefer streaming body over buffered body
	match body {
		RequestBody::Stream(byte_stream) => {
			// A body read from a stream has no length to advertise, which the fetch standard
			// allows only over HTTP/2 and HTTP/3.
			// spec:REQ#streaming-a-request-body
			if !agent.quirk_h1_request_streaming {
				// Faith never negotiates h2c, so a plaintext origin is HTTP/1.x for certain and
				// can be refused without opening a connection to find out. Returning here drops
				// the stream, which is what tells whatever is feeding it to stop.
				if parsed_url.scheme() != "https" {
					return Err(FaithError::new(
						FaithErrorKind::Network,
						format!(
							"a streaming request body requires HTTP/2 or HTTP/3, and {} is served over HTTP/1.1; set the agent's quirks.h1RequestStreaming to send it anyway",
							parsed_url.as_str()
						),
					));
				}

				// Over TLS the protocol is only known once ALPN has run. Asserting HTTP/2 on the
				// request hands the check to the layer that finds out: the connection is chosen,
				// and an HTTP/1.x one is refused there before any of the body is written.
				request = request.version(http::Version::HTTP_2);
			}

			#[cfg(feature = "encoding")]
			let body = match compress.clone() {
				// Compressed as the chunks arrive, and chunked on the wire either way:
				// a stream has no length to declare up front.
				// spec:ENC#what-a-compressed-request-sends
				Some(coding) => {
					applied_coding = Some(coding.clone());
					let stream =
						encoding_request::compress_stream(byte_stream, coding).map_err(|err| {
							FaithError::new(FaithErrorKind::InvalidCompression, err.to_string())
						})?;
					reqwest::Body::wrap_stream(stream)
				}
				None => reqwest::Body::wrap_stream(byte_stream),
			};
			#[cfg(not(feature = "encoding"))]
			let body = reqwest::Body::wrap_stream(byte_stream);
			request = request.body(body);
		}
		RequestBody::Bytes(bytes) => {
			#[cfg(feature = "encoding")]
			let body = match compress.clone() {
				// The compressed bytes are what reqwest sizes `Content-Length` from, so the
				// header counts what goes on the wire.
				// spec:ENC#what-a-compressed-request-sends
				Some(coding) => {
					applied_coding = Some(coding.clone());
					encoding_request::compress_buffer(&bytes, coding)
						.await
						.map_err(|err| {
							FaithError::new(
								FaithErrorKind::Network,
								format!("could not compress the request body: {err}"),
							)
						})?
				}
				None => bytes.to_vec(),
			};
			#[cfg(not(feature = "encoding"))]
			let body = bytes.to_vec();
			request = request.body(body);
		}
		RequestBody::None => {}
	}

	// One `Content-Encoding` naming the caller's codings then Faith's, in the order they
	// were applied (spec:ENC#what-a-compressed-request-sends).
	#[cfg(feature = "encoding")]
	if let Some(coding) = applied_coding {
		let layered = declared_content_encoding
			.as_deref()
			.map(ContentEncoding::from)
			.unwrap_or_default()
			.layer(coding);
		let value = layered.to_header_value().ok_or_else(|| {
			FaithError::new(
				FaithErrorKind::InvalidHeader,
				format!("invalid header value: {layered}"),
			)
		})?;
		request = request.header(CONTENT_ENCODING, value);
	}

	if let Some(dur) = options.timeout {
		request = request.timeout(dur);
	}

	agent.stats.requests_sent.fetch_add(1, Ordering::Relaxed);

	// The origin every phase is measured from.
	let started = Instant::now();

	// A caller that can abort races the request against that signal; one that cannot just sends.
	let response = match abort {
		Some(abort) => {
			tokio::select! {
				result = request.send() => result?,
				_ = abort => {
					return Err(FaithErrorKind::Aborted.into());
				}
			}
		}
		None => request.send().await?,
	};

	agent
		.stats
		.responses_received
		.fetch_add(1, Ordering::Relaxed);

	let status_code = response.status();
	let empty = status_code == StatusCode::NO_CONTENT || is_head;

	let response_url = response.url().clone();
	let version = response.version();

	// With `http3.upgradeFollowAdvertisedPort` on, an HTTP/3 attempt rewrites the
	// request's port to the advertised one, so the response URL's port reflects
	// which endpoint answered rather than any redirect. Compare with ports
	// normalised away, or every such request would report `redirected`.
	//
	// Only HTTP/3 responses can have been rewritten — reqwest routes
	// `Version::HTTP_3` exclusively to the h3 client with no silent downgrade, and
	// the TCP fallback re-runs the untouched clone. Restricting the normalisation
	// to those keeps exact comparison, and so port-only redirect detection, for
	// every other response.
	let redirected = if agent.h3_follow_advertised_port && version == http::Version::HTTP_3 {
		let without_port = |url: &reqwest::Url| {
			let mut url = url.clone();
			let _ = url.set_port(None);
			url
		};
		without_port(&parsed_url) != without_port(&response_url)
	} else {
		parsed_url != response_url
	};

	// Track connection for TCP stats (if we can get both local and remote addr).
	// A connection the tracker has already seen is one the pool handed back, which is
	// what `reused` reports (spec:RESP#request-timing).
	#[cfg(feature = "connection-tracking")]
	let reused = if let Some(http_info) = response.extensions().get::<HttpInfo>() {
		let local_addr = http_info.local_addr();
		let remote_addr = http_info.remote_addr();
		agent.conn_tracker.track(local_addr, remote_addr)
	} else {
		false
	};
	// Whether the pool handed a connection back is what the tracker knows, so without it there is
	// no answer to report.
	#[cfg(not(feature = "connection-tracking"))]
	let reused = false;

	// The origin now holds a connection the pool keeps idle, so a `preconnect` for it has
	// nothing left to do (spec:WARM). Keyed on the URL the request was sent to, so a
	// redirect chain marks the origin that actually answered rather than the one asked for.
	agent.mark_warm(&response_url);

	let peer = PeerInformation {
		address: response.remote_addr(),
		certificate: response
			.extensions()
			.get::<TlsInfo>()
			.and_then(|info| info.peer_certificate())
			.map(|cert| cert.into()),
	};

	let mut headers = response.headers().clone();
	if options.credentials == Credentials::Omit {
		headers.remove("set-cookie");
	}

	// Taken here rather than inside the stack: the layer that used to stamp was only built with
	// HTTP/3, and never saw a cache hit at all (spec:RESP#request-timing, Q3).
	let headers_at = Instant::now();
	let timing = RequestTiming {
		headers_ms: headers_at.duration_since(started).as_secs_f64() * 1000.0,
		body_ms: None,
		reused,
		next_hop_protocol: alpn_protocol_id(version, &response_url),
		// Captured before a decoded body's `Content-Encoding` is stripped below, so the
		// coding the response arrived under is reported either way.
		content_encoding: headers
			.get(CONTENT_ENCODING)
			.and_then(|value| value.to_str().ok())
			.map(str::to_owned),
		from_cache: headers
			.get("x-cache")
			.and_then(|value| value.to_str().ok())
			.is_some_and(|value| value.eq_ignore_ascii_case("HIT")),
	};

	// Decode only a body Faith negotiated the coding for; a bodyless response keeps its
	// `Content-Encoding` and `Content-Length` describing the representation (spec: ENC).
	#[cfg(feature = "encoding")]
	let decode = if empty {
		None
	} else {
		// The body is decoded lazily when it is read, so the header edit happens here and the
		// stream is wrapped there; `response::decode` is the one-call form for anyone whose
		// body is in hand.
		ContentEncoding::peel_one_header(&mut headers, &accept_encoding)
	};

	let timing = Arc::new(TimingSlot::new(started, timing));
	// A response that cannot carry a body has nothing left to wait for.
	if empty {
		timing.ended();
	}

	let trailers = Arc::new(TrailersSlot::default());
	let claim = (!empty).then(|| {
		let http_response: http::Response<_> = response.into();
		BodyShared::first_claim(BodyParts {
			body: http_response.into_body(),
			version,
			drain: agent.drain,
			#[cfg(feature = "encoding")]
			decode,
			trailers: trailers.clone(),
			timing: timing.clone(),
			stats: agent.stats.clone(),
		})
	});

	Ok(Response {
		claim,
		disturbed: Arc::new(AtomicBool::new(false)),
		headers,
		integrity: options.integrity,
		peer: Arc::new(peer),
		redirected,
		status_code,
		timing,
		trailers,
		url: response_url,
		version,
	})
}