vorma 0.86.0-pre.5

//! The Tower services that an application mounts: Vorma's application service and the
//! static public assets layer.

use std::convert::Infallible;
use std::error::Error as StdError;
use std::io::SeekFrom;
use std::path::Path;
use std::sync::Arc;
use std::task::{Context, Poll};

use bytes::{Buf, Bytes};
use futures_util::future::{self, BoxFuture, FutureExt};
use futures_util::{StreamExt, TryStreamExt};
use http::header::{ACCEPT_RANGES, CACHE_CONTROL, CONTENT_RANGE, CONTENT_TYPE};
use http::uri::PathAndQuery;
use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Uri};
use http_body_util::{BodyExt, Either, Empty, Limited};
use primitives_rs::tasks::{TaskCtx, TaskRuntime};
use serde_json::Value;
use tokio::io::AsyncSeekExt;
use tower_layer::Layer;
use tower_service::Service;
use vorma_path_wire::canonicalize_request_path;

use crate::build_output::{BuildOutput, BuildSnapshot, CspSources, RuntimeMode};
use crate::ctx::{MatchedApiEndpoint, MatchedPage, PatternMatch, RequestFacts, RequestMatch};
use crate::declarations::{
	ApiEndpointInputKind, ApiEndpointInputSource, InputError, Invocation, input_failure_outcome,
};
use crate::dev::READINESS_PATH;
use crate::document::HeadRules;
use crate::execution::{Committed, FailedPosition, PhaseInvocation, PhaseResult, run_phase};
use crate::finalize::{
	ACCEPTS_CLIENT_REDIRECT_HEADER, CLIENT_BUILD_ID_HEADER, ResolvedRedirect, VormaBody,
	finalize_api_endpoint_failure, finalize_api_endpoint_or_report, finalize_api_endpoint_success,
	finalize_build_skew, finalize_html, finalize_json, finalize_not_modified, finalize_or_report,
	finalize_readiness, finalize_redirect, finalize_status,
};
use crate::media_type::{FormMediaType, FormMediaTypeError, is_json_media_type};
use crate::outcome::{
	Contribution, HandlerKind, HandlerLabel, Outcome, RedirectKind, Redirection, ResolveTarget,
	invalid_input_outcome,
};
use crate::pattern::DecodedMatch;
use crate::payload::{
	CompiledPatternWire, DocumentInput, FailureWire, NavPayload, OrderedObject, RouteRecordWire,
	SsrPayload, prepare_assets, render_document, to_document_attrs_wire, to_head_wire,
};
use crate::plan::{
	ApiEndpointNode, ClassifiedRequest, CurrentPlan, Declarations, Plan, ViewPosition,
};
use crate::search::{DEPLOYMENT_QUERY_KEY, VIEW_PAYLOAD_QUERY_KEY, parse_search_params};
use crate::static_public_asset::{
	ByteRange, StaticPublicAssetAnswer, StaticPublicAssetBody, decide_static_public_asset_answer,
	read_content_type,
};

/// Vorma's application service, a Tower service. Mount it where the application
/// delegates requests to Vorma, such as an axum router's fallback service.
///
/// Mount the service so that it receives each request with its complete
/// application path, as the browser's address bar shows it. Vorma matches the
/// path against its patterns, and the browser client matches the path in the
/// address bar against the same patterns, so a mount or a host layer that
/// removes or rewrites part of the path makes the two disagree. For example,
/// axum's `Router::nest("/app", ...)` removes `/app` from the path that the
/// nested service sees, so the service would match `/app/docs` as `/docs`
/// while the browser matches it as `/app/docs`. A fallback service receives
/// the complete path.
///
/// Every handler of a request (task middleware, view loaders, and API
/// endpoints) runs in its own Tokio task, so a `tokio::task_local!` value that
/// the host sets around the call to this service is not visible to any handler.
/// A host that has per-request values puts them in the request's extensions,
/// which every handler reads from `request`.
///
/// A request that matches no Vorma pattern goes to the fallback, which
/// answers with an empty 404 unless [`AppService::set_fallback`] replaces it.
/// Vorma collects a JSON request body in memory before it decodes it and
/// streams uploaded files to temporary files. It reads at most
/// [`App::max_request_body_size`](crate::App::max_request_body_size) of a
/// body, or the limit that
/// [`ApiEndpoint::set_max_request_body_size`](crate::ApiEndpoint::set_max_request_body_size)
/// sets for the endpoint, and answers a larger body with status 413. A limit
/// of the host server applies as well, for example tower-http's
/// `RequestBodyLimitLayer`, and the lower limit wins; axum's
/// `DefaultBodyLimit` does not apply to a fallback service.
///
/// The service creates each request's task context from the task runtime that
/// the application gave to [`LoadedApp::service`](crate::LoadedApp::service).
pub struct AppService<State, Fallback = NotFound> {
	inner: Arc<AppServiceInner<State>>,
	fallback: Fallback,
}

impl<State, Fallback: Clone> Clone for AppService<State, Fallback> {
	fn clone(&self) -> Self {
		Self {
			inner: Arc::clone(&self.inner),
			fallback: self.fallback.clone(),
		}
	}
}

pub(crate) struct AppServiceInner<State> {
	pub(crate) declarations: Arc<Declarations<State>>,
	pub(crate) head_rules: Arc<HeadRules>,
	pub(crate) build: Arc<BuildOutput>,
	pub(crate) plan: Arc<CurrentPlan<State>>,
	pub(crate) max_request_body_size: usize,
	pub(crate) tasks: TaskRuntime<State>,
	pub(crate) deployment_id: Option<String>,
}

impl<State: Send + Sync + 'static> AppService<State> {
	pub(crate) fn new(inner: AppServiceInner<State>) -> Self {
		Self {
			inner: Arc::new(inner),
			fallback: NotFound,
		}
	}
}

impl<State, Fallback> AppService<State, Fallback> {
	/// Serves requests that match no Vorma pattern with this Tower service,
	/// such as a page for unknown URLs, as tower-http's `ServeDir::fallback`
	/// does.
	pub fn set_fallback<NewFallback>(
		self,
		fallback: NewFallback,
	) -> AppService<State, NewFallback> {
		AppService {
			inner: self.inner,
			fallback,
		}
	}
}

/// The default fallback of an [`AppService`]: an empty 404 response.
#[derive(Clone, Copy, Debug)]
pub struct NotFound;

impl<RequestBody> Service<Request<RequestBody>> for NotFound {
	type Response = Response<Empty<Bytes>>;
	type Error = Infallible;
	type Future = future::Ready<Result<Response<Empty<Bytes>>, Infallible>>;

	fn poll_ready(&mut self, _ctx: &mut Context<'_>) -> Poll<Result<(), Infallible>> {
		Poll::Ready(Ok(()))
	}

	fn call(&mut self, _request: Request<RequestBody>) -> Self::Future {
		let mut response = Response::new(Empty::new());
		*response.status_mut() = StatusCode::NOT_FOUND;
		future::ready(Ok(response))
	}
}

impl<State, Fallback, FallbackBody, RequestBody> Service<Request<RequestBody>>
	for AppService<State, Fallback>
where
	State: Send + Sync + 'static,
	Fallback: Service<Request<RequestBody>, Response = Response<FallbackBody>, Error = Infallible>
		+ Clone
		+ Send
		+ 'static,
	Fallback::Future: Send,
	RequestBody: http_body::Body + Send + 'static,
	RequestBody::Data: Send,
	RequestBody::Error: Into<Box<dyn StdError + Send + Sync>>,
{
	type Response = Response<Either<VormaBody, FallbackBody>>;
	type Error = Infallible;
	type Future = BoxFuture<'static, Result<Self::Response, Infallible>>;

	fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Infallible>> {
		self.fallback.poll_ready(ctx)
	}

	fn call(&mut self, request: Request<RequestBody>) -> Self::Future {
		// The ready clone serves this request; a fresh clone waits for the next.
		let clone = self.fallback.clone();
		let mut fallback = std::mem::replace(&mut self.fallback, clone);
		let inner = Arc::clone(&self.inner);
		async move {
			match inner.handle(request).await {
				Handled::Response(response) => Ok(response.map(Either::Left)),
				Handled::Unmatched(request) => fallback
					.call(request)
					.await
					.map(|response| response.map(Either::Right)),
			}
		}
		.boxed()
	}
}

/// How Vorma handled a request.
enum Handled<RequestBody> {
	Response(Response<VormaBody>),
	/// The request matches no Vorma pattern, so the fallback serves it.
	Unmatched(Request<RequestBody>),
}

/// What every invocation of one request shares.
struct RequestSession<'plan, State> {
	plan: &'plan Arc<Plan<State>>,
	facts: Arc<RequestFacts>,
	task_ctx: TaskCtx<State>,
	accepts_client_redirect: bool,
	/// The request's path, without its query, for reports.
	request_path: Arc<str>,
}

impl<State> RequestSession<'_, State> {
	/// What reports about a handler of this request name.
	fn handler_label(&self, kind: HandlerKind, pattern: Arc<str>) -> HandlerLabel {
		HandlerLabel {
			kind,
			pattern,
			request_path: Arc::clone(&self.request_path),
		}
	}

	fn client_build_id(&self) -> &str {
		&self.plan.snapshot.metadata.client_build_id
	}

	/// A navigation redirect is a client redirect when the request accepts
	/// one; an endpoint redirect is always a 3xx response.
	fn finalize_redirect(
		&self,
		redirection: Redirection,
		contribution: &Contribution,
	) -> Response<VormaBody> {
		finalize_or_report(
			finalize_redirect(
				ResolvedRedirect {
					is_followed_by_client: self.accepts_client_redirect
						&& redirection.kind == RedirectKind::Nav,
					location: redirection.location,
					status: redirection.status,
				},
				contribution,
				self.client_build_id(),
			),
			self.client_build_id(),
		)
	}
}

impl<State: Send + Sync + 'static> AppServiceInner<State> {
	/// The plan for the current build. In development, a new build gets a new
	/// plan when a request first sees it. `None` after a build that cannot be
	/// read or planned, which the read or the planning has reported.
	async fn read_plan(&self) -> Option<Arc<Plan<State>>> {
		let snapshot = self.build.read_snapshot().await.ok()?;
		self.plan.update(&self.declarations, snapshot).ok()
	}

	fn spawn_invocation<Input>(
		&self,
		session: &RequestSession<'_, State>,
		invoke_handler_fn: &(
		     dyn Fn(Invocation<State, Input>) -> BoxFuture<'static, Outcome> + Send + Sync
		 ),
		decoded_match: DecodedMatch,
		input: Input,
		label: HandlerLabel,
	) -> PhaseInvocation {
		let future = invoke_handler_fn(Invocation {
			state: Arc::clone(self.tasks.state()),
			facts: Arc::clone(&session.facts),
			task_ctx: session.task_ctx.clone(),
			decoded_match,
			input,
			label: label.clone(),
		});
		PhaseInvocation { future, label }
	}

	async fn handle<RequestBody>(&self, request: Request<RequestBody>) -> Handled<RequestBody>
	where
		RequestBody: http_body::Body + Send + 'static,
		RequestBody::Data: Send,
		RequestBody::Error: Into<Box<dyn StdError + Send + Sync>>,
	{
		let Some(plan) = self.read_plan().await else {
			return Handled::Response(finalize_status(StatusCode::INTERNAL_SERVER_ERROR, None));
		};
		let snapshot: &BuildSnapshot = &plan.snapshot;
		let client_build_id = snapshot.metadata.client_build_id.as_str();
		let is_dev = snapshot.metadata.runtime_mode == RuntimeMode::Dev;
		let (mut parts, body) = request.into_parts();
		if is_dev && parts.method == Method::GET && parts.uri.path() == READINESS_PATH {
			return Handled::Response(finalize_readiness(client_build_id));
		}
		let bad_request = || {
			Handled::Response(finalize_status(
				StatusCode::BAD_REQUEST,
				Some(client_build_id),
			))
		};
		let Ok(path) = canonicalize_request_path(parts.uri.path()) else {
			return bad_request();
		};
		let mut search_entries: Vec<(String, String)> =
			form_urlencoded::parse(parts.uri.query().unwrap_or_default().as_bytes())
				.into_owned()
				.collect();
		// A HEAD request is answered as the GET request for its URL; the
		// server omits the body.
		let method = if parts.method == Method::HEAD {
			Method::GET
		} else {
			parts.method.clone()
		};
		let target = match plan.classify(&method, &path) {
			Ok(ClassifiedRequest::Unmatched) => {
				return Handled::Unmatched(Request::from_parts(parts, body));
			}
			Ok(target) => target,
			Err(_) => return bad_request(),
		};
		let is_view = matches!(target, ClassifiedRequest::View { .. });
		let submitted: Vec<String> = if is_view {
			search_entries
				.iter()
				.filter(|(key, _)| key == VIEW_PAYLOAD_QUERY_KEY)
				.map(|(_, value)| value.clone())
				.collect()
		} else {
			let mut values = Vec::new();
			for value in parts.headers.get_all(CLIENT_BUILD_ID_HEADER) {
				let Ok(value) = value.to_str() else {
					return bad_request();
				};
				values.push(value.to_owned());
			}
			values
		};
		if submitted.len() > 1 {
			return bad_request();
		}
		let submitted = submitted.into_iter().next();
		let is_json_request = submitted.as_deref().is_some_and(|id| !id.is_empty());
		// A navigation payload is only served to the build that can render it. An
		// endpoint request from another build executes normally; its response
		// carries this build's ID, which the browser reports as a mismatch.
		let names_other_build = submitted
			.as_deref()
			.is_some_and(|id| id != client_build_id && id != "_");
		if is_view && is_json_request && names_other_build {
			return Handled::Response(finalize_build_skew(client_build_id));
		}
		let skew_protection = self.deployment_id.is_some();
		if is_view {
			search_entries.retain(|(key, _)| !is_transport_param(key, skew_protection));
		}
		let accepts_client_redirect = parts.headers.contains_key(ACCEPTS_CLIENT_REDIRECT_HEADER);
		let target_resolver: Arc<dyn ResolveTarget> = Arc::clone(&plan) as Arc<dyn ResolveTarget>;
		let request_match = match &target {
			ClassifiedRequest::ApiEndpoint {
				node,
				decoded_match,
			} => RequestMatch::ApiEndpoint(MatchedApiEndpoint {
				kind: node.declaration.kind,
				method: node.declaration.method.clone(),
				pattern: node.declaration.pattern,
				decoded_match: decoded_match.clone(),
			}),
			ClassifiedRequest::View { positions } => RequestMatch::Page(MatchedPage {
				pattern_matches: positions
					.iter()
					.map(|position| PatternMatch {
						pattern: position.node.pattern.clone(),
						decoded_match: position.decoded_match.clone(),
					})
					.collect(),
			}),
			ClassifiedRequest::Unmatched => unreachable!("an unmatched request returned earlier"),
		};
		if is_view {
			parts.uri = without_transport_params(&parts.uri, skew_protection);
		}
		let request_path: Arc<str> = Arc::from(parts.uri.path());
		let facts = Arc::new(RequestFacts {
			target_resolver,
			head_rules: is_view.then(|| Arc::clone(&self.head_rules)),
			request_match,
			request_parts: Arc::new(parts),
			latest_known_client_build_id: client_build_id.to_owned(),
			// The placeholder `_` skips the skew check and names no build.
			requested_client_build_id: submitted.filter(|id| is_json_request && id != "_"),
			static_public_assets: Arc::clone(&snapshot.static_public_assets),
			csp_sources: if is_view {
				Arc::clone(&snapshot.page_csp_sources)
			} else {
				Arc::new(CspSources::default())
			},
		});
		let session = RequestSession {
			plan: &plan,
			facts,
			task_ctx: self.tasks.task_ctx(),
			accepts_client_redirect,
			request_path,
		};
		let middleware = plan
			.select_middleware(&method, &path)
			.into_iter()
			.map(|node| {
				let invocation = self.spawn_invocation(
					&session,
					&*node.declaration.invoke_handler_fn,
					DecodedMatch::default(),
					(),
					session.handler_label(HandlerKind::TaskMiddleware, Arc::clone(&node.pattern)),
				);
				((), Some(invocation))
			})
			.collect();
		let middleware = run_phase(middleware, Contribution::default()).await;
		Handled::Response(match target {
			ClassifiedRequest::ApiEndpoint {
				node,
				decoded_match,
			} => {
				self.respond_api_endpoint(ApiEndpointRespondInput {
					session: &session,
					middleware,
					node,
					decoded_match,
					search_entries: &search_entries,
					body,
				})
				.await
			}
			ClassifiedRequest::View { positions } => {
				self.respond_view(ViewRespondInput {
					session: &session,
					middleware,
					positions,
					search_entries: &search_entries,
					is_json_request,
				})
				.await
			}
			ClassifiedRequest::Unmatched => unreachable!("an unmatched request returned earlier"),
		})
	}

	async fn respond_api_endpoint<RequestBody>(
		&self,
		request: ApiEndpointRespondInput<'_, '_, State, RequestBody>,
	) -> Response<VormaBody>
	where
		RequestBody: http_body::Body + Send + 'static,
		RequestBody::Data: Send,
		RequestBody::Error: Into<Box<dyn StdError + Send + Sync>>,
	{
		let session = request.session;
		let client_build_id = session.client_build_id();
		let contribution = match request.middleware {
			PhaseResult::Completed { contribution, .. } => contribution,
			PhaseResult::Redirect {
				contribution,
				redirection,
			} => return session.finalize_redirect(redirection, &contribution),
			PhaseResult::Failed {
				contribution,
				failed,
				..
			} => {
				return finalize_api_endpoint_or_report(
					finalize_api_endpoint_failure(
						failed.status,
						failed.public_error.as_ref(),
						&contribution,
						client_build_id,
					),
					client_build_id,
				);
			}
		};
		let node: &ApiEndpointNode<State> = request.node;
		let body_limit = node
			.declaration
			.max_request_body_size
			.unwrap_or(self.max_request_body_size);
		let label = session.handler_label(
			HandlerKind::ApiEndpoint,
			Arc::from(node.declaration.pattern),
		);
		let source = match node.input_kind {
			ApiEndpointInputKind::None => Ok(ApiEndpointInputSource::None),
			ApiEndpointInputKind::Search => {
				let shape = node
					.search
					.as_ref()
					.expect("a search endpoint has a search shape");
				parse_search_params(shape, request.search_entries)
					.map(ApiEndpointInputSource::Search)
					.map_err(|_| InputError::Invalid)
			}
			ApiEndpointInputKind::Json => {
				if session
					.facts
					.request_parts
					.headers
					.get(CONTENT_TYPE)
					.is_some_and(is_json_media_type)
				{
					match Limited::new(request.body, body_limit).collect().await {
						Ok(collected) => Ok(ApiEndpointInputSource::Json(collected.to_bytes())),
						Err(error) => Err(InputError::BodyRead(error)),
					}
				} else {
					Err(InputError::UnsupportedMediaType)
				}
			}
			// A form is read while it is decoded, so uploaded files stream to
			// temporary files instead of memory.
			ApiEndpointInputKind::FormData => {
				let media_type = session
					.facts
					.request_parts
					.headers
					.get(CONTENT_TYPE)
					.map(FormMediaType::parse);
				match media_type {
					Some(Ok(media_type)) => Ok(ApiEndpointInputSource::FormData {
						media_type,
						body: Limited::new(request.body, body_limit)
							.into_data_stream()
							.map_ok(|mut data| data.copy_to_bytes(data.remaining()))
							.map_err(Into::into)
							.boxed(),
					}),
					// A multipart form without a boundary is a form that cannot be read.
					Some(Err(FormMediaTypeError::NoBoundary)) => Err(InputError::Invalid),
					Some(Err(FormMediaTypeError::NotAForm)) | None => {
						Err(InputError::UnsupportedMediaType)
					}
				}
			}
		};
		let invocation = match source {
			Ok(source) => self.spawn_invocation(
				session,
				&*node.declaration.invoke_handler_fn,
				request.decoded_match,
				source,
				label,
			),
			Err(error) => PhaseInvocation {
				future: future::ready(input_failure_outcome(&label, error)).boxed(),
				label,
			},
		};
		let response = match run_phase(vec![((), Some(invocation))], contribution).await {
			PhaseResult::Completed {
				contribution,
				positions,
			} => {
				let committed = positions.into_iter().next();
				match committed {
					Some(Committed::NotModified { .. }) => {
						finalize_not_modified(&contribution, client_build_id)
					}
					Some(Committed::Completed { output, .. }) => finalize_api_endpoint_success(
						output.as_deref(),
						&contribution,
						client_build_id,
					),
					Some(Committed::NoHandler { .. }) | None => {
						finalize_api_endpoint_success(None, &contribution, client_build_id)
					}
				}
			}
			PhaseResult::Redirect {
				contribution,
				redirection,
			} => return session.finalize_redirect(redirection, &contribution),
			PhaseResult::Failed {
				contribution,
				failed,
				..
			} => finalize_api_endpoint_failure(
				failed.status,
				failed.public_error.as_ref(),
				&contribution,
				client_build_id,
			),
		};
		finalize_api_endpoint_or_report(response, client_build_id)
	}

	async fn respond_view(
		&self,
		request: ViewRespondInput<'_, '_, '_, State>,
	) -> Response<VormaBody> {
		let session = request.session;
		// Each loader's search is parsed once: its decoder reads it here and
		// the payload carries it to the browser.
		let searches: Vec<Option<Result<Value, ()>>> = request
			.positions
			.iter()
			.map(|position| {
				let search = position.node.loader.as_ref()?.search.as_ref()?;
				Some(parse_search_params(search, request.search_entries).map_err(|_| ()))
			})
			.collect();
		let contribution = match request.middleware {
			PhaseResult::Completed { contribution, .. } => contribution,
			PhaseResult::Redirect {
				contribution,
				redirection,
			} => return session.finalize_redirect(redirection, &contribution),
			// A rejected page request renders the application boundary: the
			// first position fails with the rejection, and no position renders.
			PhaseResult::Failed {
				contribution,
				failed,
				..
			} => {
				return self.render_view(RenderViewInput {
					session,
					positions: &request.positions,
					searches: &searches,
					committed: Vec::new(),
					failed_position: Some(FailedPosition { index: 0, ..failed }),
					contribution,
					is_json_request: request.is_json_request,
				});
			}
		};
		let mut phase = Vec::with_capacity(request.positions.len());
		for (index, position) in request.positions.iter().enumerate() {
			let Some(loader) = &position.node.loader else {
				phase.push((index, None));
				continue;
			};
			let label = session.handler_label(
				HandlerKind::ViewLoader,
				Arc::from(loader.declaration.pattern),
			);
			let invocation = match &searches[index] {
				Some(Err(())) => PhaseInvocation {
					future: future::ready(invalid_input_outcome()).boxed(),
					label,
				},
				search => {
					let input = match search {
						Some(Ok(value)) => value.clone(),
						_ => Value::Null,
					};
					self.spawn_invocation(
						session,
						&*loader.declaration.invoke_handler_fn,
						position.decoded_match.clone(),
						input,
						label,
					)
				}
			};
			phase.push((index, Some(invocation)));
		}
		let (committed, failed_position, contribution) = match run_phase(phase, contribution).await
		{
			PhaseResult::Redirect {
				contribution,
				redirection,
			} => return session.finalize_redirect(redirection, &contribution),
			PhaseResult::Completed {
				contribution,
				positions,
			} => (positions, None, contribution),
			PhaseResult::Failed {
				contribution,
				before,
				failed,
				..
			} => (before, Some(failed), contribution),
		};
		self.render_view(RenderViewInput {
			session,
			positions: &request.positions,
			searches: &searches,
			committed,
			failed_position,
			contribution,
			is_json_request: request.is_json_request,
		})
	}

	fn render_view(&self, input: RenderViewInput<'_, '_, '_, State>) -> Response<VormaBody> {
		let session = input.session;
		let metadata = &session.plan.snapshot.metadata;
		let client_build_id = metadata.client_build_id.as_str();
		let mut routes = Vec::with_capacity(input.committed.len());
		let mut views = Vec::with_capacity(input.committed.len());
		for committed in &input.committed {
			let (index, output) = match committed {
				Committed::Completed {
					position_key,
					output,
				} => (*position_key, output.as_deref()),
				Committed::NotModified { position_key } | Committed::NoHandler { position_key } => {
					(*position_key, None)
				}
			};
			let position = &input.positions[index];
			let node = position.node;
			if let Some(module) = &node.client_view_module {
				views.push(module);
			}
			let registered = node.compiled_pattern.as_pattern().as_str();
			routes.push(RouteRecordWire {
				pattern: &node.pattern,
				params: OrderedObject(
					position
						.decoded_match
						.params
						.iter()
						.map(|(name, value)| (name.as_str(), value.as_str()))
						.collect(),
				),
				splat_values: &position.decoded_match.splat_values,
				compiled_pattern: CompiledPatternWire {
					pattern: (registered != node.pattern).then_some(registered),
					segments: node.compiled_pattern.as_pattern().segments(),
				},
				search_shape: node
					.loader
					.as_ref()
					.and_then(|loader| loader.search.as_ref()),
				search_input: match &input.searches[index] {
					Some(Ok(value)) => Some(value),
					_ => None,
				},
				import_url: node
					.client_view_module
					.as_ref()
					.map(|module| module.import_url.as_str()),
				loader_output: output,
			});
		}
		let failure = input.failed_position.as_ref().map(|error| FailureWire {
			index: Some(error.index),
			status: error.status.as_u16(),
			public_error: error.public_error.as_ref(),
		});
		let contribution = &input.contribution;
		let document = &contribution.document;
		let head = self.head_rules.deduplicate(document.head_changes());
		let assets = prepare_assets(&metadata.client_entry, views.iter().copied());
		let response = if input.is_json_request {
			let payload = NavPayload {
				routes,
				failure,
				head: to_head_wire(&head),
				document_attrs: to_document_attrs_wire(document),
				deps: &assets.nav_dep_urls,
				css_bundles: &assets.nav_css_bundle_urls,
				referenced_static_public_asset_urls: &assets
					.nav_referenced_static_public_asset_urls,
			};
			serde_json::to_string(&payload)
				.map_err(crate::finalize::FinalizeError::Serialization)
				.and_then(|json| finalize_json(json, contribution, client_build_id))
		} else {
			let module_preload_urls: Vec<&str> =
				std::iter::once(metadata.client_entry.import_url.as_str())
					.chain(routes.iter().filter_map(|route| route.import_url))
					.chain(assets.document_dep_urls.iter().map(String::as_str))
					.collect();
			let payload = SsrPayload {
				routes,
				failure,
				client_build_id,
				document_attrs: to_document_attrs_wire(document),
				is_dev: metadata.runtime_mode == RuntimeMode::Dev,
				deployment_id: self.deployment_id.as_deref(),
			};
			serde_json::to_string(&payload)
				.map_err(crate::finalize::FinalizeError::Serialization)
				.and_then(|payload_json| {
					let html = render_document(DocumentInput {
						html_attrs: &document.html_attrs,
						head_attrs: &document.head_attrs,
						body_attrs: &document.body_attrs,
						head: &head,
						critical_css: &metadata.critical_css,
						dev_head_html: metadata.dev_head_html.as_deref(),
						dev_body_prepend_html: metadata.dev_body_prepend_html.as_deref(),
						dev_body_html: metadata.dev_body_html.as_deref(),
						css_bundle_urls: &assets.document_css_bundle_urls,
						module_preload_urls,
						referenced_static_public_asset_urls: &assets
							.document_referenced_static_public_asset_urls,
						entry_import_url: &metadata.client_entry.import_url,
						payload_json: &payload_json,
					});
					finalize_html(html, contribution, client_build_id)
				})
		};
		finalize_or_report(response, client_build_id)
	}
}

/// Whether a query key belongs to the transport: `vorma-json` always, and
/// `dpl` while Vercel skew protection is on.
fn is_transport_param(key: &str, skew_protection: bool) -> bool {
	key == VIEW_PAYLOAD_QUERY_KEY || (skew_protection && key == DEPLOYMENT_QUERY_KEY)
}

/// The URI without the transport parameters, with the rest of the query kept
/// as it arrived.
fn without_transport_params(uri: &Uri, skew_protection: bool) -> Uri {
	let Some(query) = uri.query() else {
		return uri.clone();
	};
	let kept: Vec<&str> = query
		.split('&')
		.filter(|pair| {
			let key = form_urlencoded::parse(pair.as_bytes()).next();
			!key.is_some_and(|(key, _)| is_transport_param(&key, skew_protection))
		})
		.collect();
	let path = uri.path();
	let path_and_query = if kept.is_empty() {
		path.to_owned()
	} else {
		format!("{path}?{}", kept.join("&"))
	};
	let mut uri_parts = uri.clone().into_parts();
	uri_parts.path_and_query = Some(
		path_and_query
			.parse::<PathAndQuery>()
			.expect("a path and a subset of its query are a valid path and query"),
	);
	Uri::from_parts(uri_parts).expect("replacing the path and query of a valid URI keeps it valid")
}

struct ApiEndpointRespondInput<'session, 'plan, State, RequestBody> {
	session: &'session RequestSession<'plan, State>,
	middleware: PhaseResult<()>,
	node: &'plan ApiEndpointNode<State>,
	decoded_match: DecodedMatch,
	search_entries: &'session [(String, String)],
	body: RequestBody,
}

struct ViewRespondInput<'session, 'plan, 'entries, State> {
	session: &'session RequestSession<'plan, State>,
	middleware: PhaseResult<()>,
	positions: Vec<ViewPosition<'plan, State>>,
	search_entries: &'entries [(String, String)],
	is_json_request: bool,
}

struct RenderViewInput<'session, 'plan, 'request, State> {
	session: &'session RequestSession<'plan, State>,
	positions: &'request [ViewPosition<'plan, State>],
	searches: &'request [Option<Result<Value, ()>>],
	committed: Vec<Committed<usize>>,
	failed_position: Option<FailedPosition>,
	contribution: Contribution,
	is_json_request: bool,
}

/// Serves each published static public asset and Vorma's browser modules
/// with their published cache policy, and passes every other request on.
/// Without it, or another server for the build directory's `public`
/// directory, every request for the application's JavaScript gets a 404. A
/// host that serves that directory through a CDN or another file server does
/// not install it.
///
/// The layer answers `GET` and `HEAD` requests for a file as Go's
/// `http.FileServer` does for ranges, so that browsers can play published
/// video and audio:
///
/// - Every `200`, `206`, and `416` answer carries `Accept-Ranges: bytes` and
///   the immutable `Cache-Control`. No answer carries `Last-Modified` or
///   `ETag`, because a published file is immutable under its URL.
/// - A `Range` header with one satisfiable range gets `206` with
///   `Content-Range`, and the layer streams only the selected bytes. A
///   `Range` header that selects no byte of a non-empty file, or that is not
///   a list of byte ranges, gets `416` with `Content-Range: bytes */<length>`.
///   An empty file ignores `Range`.
/// - A `Range` header with several satisfiable ranges gets `200` with the
///   whole file, which RFC 9110 permits. The layer never answers with
///   `multipart/byteranges`.
/// - A request with `If-Range` gets `200` with the whole file, because the
///   file has no validator to compare. Conditional headers other than
///   `If-Range` have no effect.
/// - A `HEAD` request gets the status and headers of the `GET` request.
#[derive(Clone)]
pub struct StaticPublicAssetsLayer {
	build: Arc<BuildOutput>,
}

impl StaticPublicAssetsLayer {
	pub(crate) fn new(build: Arc<BuildOutput>) -> Self {
		Self { build }
	}
}

impl<Inner> Layer<Inner> for StaticPublicAssetsLayer {
	type Service = StaticPublicAssetsService<Inner>;

	fn layer(&self, inner: Inner) -> Self::Service {
		StaticPublicAssetsService {
			inner,
			build: Arc::clone(&self.build),
		}
	}
}

/// The service that [`StaticPublicAssetsLayer`] makes.
#[derive(Clone)]
pub struct StaticPublicAssetsService<Inner> {
	inner: Inner,
	build: Arc<BuildOutput>,
}

const IMMUTABLE_CACHE_CONTROL: &str = "public, max-age=31536000, immutable";

impl<Inner, RequestBody, ResponseBody> Service<Request<RequestBody>>
	for StaticPublicAssetsService<Inner>
where
	Inner:
		Service<Request<RequestBody>, Response = Response<ResponseBody>> + Clone + Send + 'static,
	Inner::Future: Send,
	RequestBody: Send + 'static,
{
	type Response = Response<Either<StaticPublicAssetBody, ResponseBody>>;
	type Error = Inner::Error;
	type Future = BoxFuture<'static, Result<Self::Response, Inner::Error>>;

	fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Inner::Error>> {
		self.inner.poll_ready(ctx)
	}

	fn call(&mut self, request: Request<RequestBody>) -> Self::Future {
		// The ready clone serves this request; a fresh clone waits for the next.
		let clone = self.inner.clone();
		let mut inner = std::mem::replace(&mut self.inner, clone);
		let build = Arc::clone(&self.build);
		let method = request.method().clone();
		let path = request.uri().path().to_owned();
		let request_headers = request.headers().clone();
		async move {
			if let Some(response) =
				serve_static_public_asset(&build, &method, &request_headers, &path).await
			{
				return Ok(response.map(Either::Left));
			}
			inner
				.call(request)
				.await
				.map(|response| response.map(Either::Right))
		}
		.boxed()
	}
}

/// A `GET` or `HEAD` request with a raw path that is exactly a published URL.
///
/// The answer follows `http.FileServer`'s ranges: `Accept-Ranges`, `206` for
/// one satisfiable range, and `416` for a range that selects nothing. See
/// [`decide_static_public_asset_answer`].
async fn serve_static_public_asset(
	build: &BuildOutput,
	method: &Method,
	request_headers: &HeaderMap,
	path: &str,
) -> Option<Response<StaticPublicAssetBody>> {
	if method != Method::GET && method != Method::HEAD {
		return None;
	}
	let Ok(snapshot) = build.read_snapshot().await else {
		return Some(create_status_response(StatusCode::INTERNAL_SERVER_ERROR).await);
	};
	let file_name = snapshot.static_public_asset_file_names_by_url.get(path)?;
	let path = build.public_dir().join(file_name);
	let answer = match answer_static_public_asset_request(&path, method, request_headers).await {
		Ok(answer) => answer,
		Err(error) => {
			tracing::error!(path = %path.display(), error = &error as &(dyn StdError + 'static), "Vorma could not read a static public asset");
			return Some(create_status_response(StatusCode::INTERNAL_SERVER_ERROR).await);
		}
	};
	let mut response = Response::new(answer.body);
	*response.status_mut() = answer.status;
	let headers = response.headers_mut();
	headers.insert(
		CACHE_CONTROL,
		HeaderValue::from_static(IMMUTABLE_CACHE_CONTROL),
	);
	headers.insert(ACCEPT_RANGES, HeaderValue::from_static("bytes"));
	if matches!(answer.status, StatusCode::OK | StatusCode::PARTIAL_CONTENT)
		&& let Ok(content_type) = HeaderValue::from_str(&read_content_type(file_name))
	{
		headers.insert(CONTENT_TYPE, content_type);
	}
	if let Some(content_range) = answer.content_range
		&& let Ok(value) = HeaderValue::from_str(&content_range)
	{
		headers.insert(CONTENT_RANGE, value);
	}
	Some(response)
}

/// The status, body, and range header of the answer to a request
/// for the file at `path`.
struct StaticPublicAssetResponseParts {
	status: StatusCode,
	body: StaticPublicAssetBody,
	content_range: Option<String>,
}

/// Opens the file and decides the answer. The body of a `GET` answer streams
/// only the selected bytes, and the body of a `HEAD` answer reports their
/// length without opening a stream.
async fn answer_static_public_asset_request(
	path: &Path,
	method: &Method,
	request_headers: &HeaderMap,
) -> std::io::Result<StaticPublicAssetResponseParts> {
	let mut file = tokio::fs::File::open(path).await?;
	let metadata = file.metadata().await?;
	let length = metadata.len();
	let (status, selected, content_range) =
		match decide_static_public_asset_answer(request_headers, length) {
			StaticPublicAssetAnswer::Whole => {
				(StatusCode::OK, ByteRange { start: 0, length }, None)
			}
			StaticPublicAssetAnswer::Partial(range) => (
				StatusCode::PARTIAL_CONTENT,
				range,
				Some(format!(
					"bytes {}-{}/{length}",
					range.start,
					range.start + range.length - 1
				)),
			),
			StaticPublicAssetAnswer::RangeNotSatisfiable => (
				StatusCode::RANGE_NOT_SATISFIABLE,
				ByteRange {
					start: 0,
					length: 0,
				},
				Some(format!("bytes */{length}")),
			),
		};
	let body = if method == Method::HEAD || selected.length == 0 {
		StaticPublicAssetBody::from_length_only(selected.length)
	} else {
		file.seek(SeekFrom::Start(selected.start)).await?;
		StaticPublicAssetBody::from_file(file, selected.length)
	};
	Ok(StaticPublicAssetResponseParts {
		status,
		body,
		content_range,
	})
}

/// The status response that Vorma makes, with its body in memory.
async fn create_status_response(status: StatusCode) -> Response<StaticPublicAssetBody> {
	let (parts, body) = finalize_status(status, None).into_parts();
	let bytes = match body.collect().await {
		Ok(collected) => collected.to_bytes(),
		Err(never) => match never {},
	};
	Response::from_parts(parts, StaticPublicAssetBody::from_bytes(bytes))
}