Skip to main content

webx_host/
context.rs

1//! HttpContext — runtime implementation of IHttpContext, IHttpRequest, IHttpResponse.
2//!
3//! # Request body
4//!
5//! The body is *not* read eagerly. It stays as the live hyper stream until
6//! something asks for it:
7//!
8//! * [`IHttpRequest::body_bytes`] buffers the whole body, enforcing
9//!   `App.MaxBodySize`.
10//! * [`IHttpRequest::multipart`] streams the body through a multipart parser,
11//!   enforcing `Form.MaxRequestSize`, `Form.MaxFileSize` and
12//!   `Form.MaxFieldSize`, and spooling large file parts to disk.
13//!
14//! Either way the body can only be consumed once; asking for it twice reports a
15//! clear error rather than silently returning an empty body.
16//!
17//! # Response body
18//!
19//! Responses carry either in-memory bytes or a [`FileBody`]. File bodies are
20//! streamed by the host with `Content-Length`, `ETag`, `Last-Modified`,
21//! `Accept-Ranges` and `Content-Disposition`, and satisfy `Range`,
22//! `If-Range`, `If-None-Match` and `If-Modified-Since` requests, so downloads
23//! can be cached and resumed.
24
25use futures_util::{StreamExt, TryStreamExt};
26use http_body_util::{combinators::UnsyncBoxBody, BodyExt, Full};
27use hyper::body::{Bytes, Frame, Incoming};
28use hyper::Request;
29use std::collections::HashMap;
30use std::net::SocketAddr;
31use std::path::{Path, PathBuf};
32use std::sync::Arc;
33use std::time::{SystemTime, UNIX_EPOCH};
34use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, SeekFrom};
35use tokio::sync::Mutex;
36use webx_core::auth::IClaims;
37use webx_core::config::AppOptions;
38use webx_core::error::{Error, Result};
39use webx_core::form::{FormFileBuilder, MultipartForm};
40use webx_core::http::{
41    content_disposition_value, ByteRange, FileBody, FileSource, IClaimsExt, IHttpContext,
42    IHttpRequest, IHttpResponse, ResponseBody, SeekableReader,
43};
44
45use crate::problem_response::{build_problem, problem_to_bytes};
46
47/// Response body produced by the HTTP host.
48///
49/// Deliberately the *unsync* box: a streamed body only ever needs to be `Send`,
50/// and requiring `Sync` would rule out readers that are not shared between
51/// threads (most streaming sources).
52pub type RespBody = UnsyncBoxBody<Bytes, std::io::Error>;
53
54/// Limits applied while reading a request body.
55///
56/// Built from `App` and `Form` configuration; see [`BodyLimits::from_options`].
57#[derive(Debug, Clone)]
58pub struct BodyLimits {
59    /// Limit for non-multipart bodies (`App.MaxBodySize`).
60    pub body: usize,
61    /// Limit for the whole multipart body (`Form.MaxRequestSize`).
62    pub multipart_body: u64,
63    /// Limit for one uploaded file (`Form.MaxFileSize`).
64    pub file: u64,
65    /// Limit for one text field (`Form.MaxFieldSize`).
66    pub field: usize,
67    /// Bytes buffered in memory per file before spooling (`Form.MemoryThreshold`).
68    pub memory_threshold: usize,
69    /// Directory for spooled uploads (`Form.TempDir`).
70    pub spool_dir: PathBuf,
71}
72
73impl Default for BodyLimits {
74    fn default() -> Self {
75        Self {
76            body: 10 * 1024 * 1024,
77            multipart_body: 256 * 1024 * 1024,
78            file: 128 * 1024 * 1024,
79            field: 1024 * 1024,
80            memory_threshold: 1024 * 1024,
81            spool_dir: webx_core::form::spool_root(),
82        }
83    }
84}
85
86impl BodyLimits {
87    /// Derive limits from the merged application configuration.
88    ///
89    /// A relative `Form.TempDir` is resolved against the application base
90    /// directory, matching how `wwwroot` and `appsettings.json` are located.
91    pub fn from_options(options: &AppOptions) -> Self {
92        let spool_dir = match options.form.temp_dir.as_deref() {
93            Some(dir) if !dir.trim().is_empty() => {
94                let path = PathBuf::from(dir);
95                if path.is_absolute() {
96                    path
97                } else {
98                    webx_core::paths::app_base().join(path)
99                }
100            }
101            _ => webx_core::form::spool_root(),
102        };
103
104        Self {
105            body: options.app.max_body_size,
106            multipart_body: options.form.max_request_size as u64,
107            file: options.form.max_file_size,
108            field: options.form.max_field_size,
109            memory_threshold: options.form.memory_threshold,
110            spool_dir,
111        }
112    }
113
114    /// Publish the spool directory so `FormFile` values can be validated
115    /// wherever they are bound.
116    pub fn apply_spool_root(&self) {
117        webx_core::form::set_spool_root(self.spool_dir.clone());
118    }
119}
120
121/// Concrete implementation of IHttpContext wrapping a hyper request and response.
122pub struct HttpContext {
123    req: HttpRequest,
124    resp: HttpResponse,
125    /// Authentication claims set by authentication middleware.
126    claims: Option<Box<dyn IClaims>>,
127    /// A request rejection recorded at construction (e.g. an oversized declared
128    /// `Content-Length`). The pipeline turns it into a normal error response
129    /// from the innermost handler so CORS/security headers/logging still run.
130    early_error: Option<Error>,
131    /// Host DI handles (provider + handler registry) for endpoint dispatch.
132    dispatch_runtime: Option<Arc<webx_core::DispatchRuntime>>,
133}
134
135impl HttpContext {
136    /// Create a context around `req`.
137    ///
138    /// No I/O happens here: the body stays a live stream until the endpoint
139    /// asks for it. `remote_addr` is the peer address captured by the accept
140    /// loop; `None` when the host cannot report it.
141    pub fn new(
142        req: Request<Incoming>,
143        limits: BodyLimits,
144        remote_addr: Option<SocketAddr>,
145        dispatch_runtime: Arc<webx_core::DispatchRuntime>,
146    ) -> Self {
147        let (parts, body) = req.into_parts();
148
149        let mut headers = HashMap::new();
150        for (name, value) in parts.headers.iter() {
151            if let Ok(value) = value.to_str() {
152                headers.insert(name.to_string(), value.to_string());
153            }
154        }
155
156        let (query_params, query_multi) = parts
157            .uri
158            .query()
159            .map(parse_query_string)
160            .unwrap_or_default();
161
162        let resp = HttpResponse::new(200);
163
164        // Refuse an oversized upload from its declared length, without reading
165        // it. Chunked bodies have no `Content-Length` and are bounded while
166        // streaming instead.
167        //
168        // The rejection is recorded, not written eagerly: writing a response
169        // here used to bypass the whole middleware pipeline (no CORS, no
170        // security headers, no access log, no metrics). The router handler
171        // turns it into a normal `PayloadTooLarge` error on the way in.
172        let declared_limit = if is_multipart_content_type(headers.get("content-type")) {
173            limits.multipart_body
174        } else {
175            limits.body as u64
176        };
177        let early_error = headers
178            .get("content-length")
179            .and_then(|value| value.parse::<u64>().ok())
180            .filter(|declared| *declared > declared_limit)
181            .map(|_| {
182                Error::PayloadTooLarge(format!(
183                    "request body exceeds the {declared_limit}-byte limit"
184                ))
185            });
186
187        let req = HttpRequest {
188            method: parts.method.to_string(),
189            path: parts.uri.path().to_string(),
190            headers,
191            query_params,
192            query_multi,
193            route_params: HashMap::new(),
194            route_pattern: None,
195            remote_addr,
196            body: Mutex::new(RequestBody::Pending(Some(body))),
197            limits,
198        };
199
200        Self {
201            req,
202            resp,
203            claims: None,
204            early_error,
205            dispatch_runtime: Some(dispatch_runtime),
206        }
207    }
208
209    /// Finish the request and produce the hyper response.
210    pub async fn into_response(self) -> hyper::Response<RespBody> {
211        let HttpContext { req, resp, .. } = self;
212        finalize_response(&req, resp).await
213    }
214}
215
216impl IClaimsExt for HttpContext {
217    fn set_claims(&mut self, claims: Box<dyn IClaims>) {
218        self.claims = Some(claims);
219    }
220
221    fn claims(&self) -> Option<&dyn IClaims> {
222        self.claims.as_deref()
223    }
224}
225
226impl IHttpContext for HttpContext {
227    fn request(&self) -> &dyn IHttpRequest {
228        &self.req
229    }
230
231    fn request_mut(&mut self) -> &mut dyn IHttpRequest {
232        &mut self.req
233    }
234
235    fn response(&self) -> &dyn IHttpResponse {
236        &self.resp
237    }
238
239    fn response_mut(&mut self) -> &mut dyn IHttpResponse {
240        &mut self.resp
241    }
242
243    fn take_early_error(&mut self) -> Option<Error> {
244        self.early_error.take()
245    }
246
247    fn dispatch_runtime(&self) -> Option<&Arc<webx_core::DispatchRuntime>> {
248        self.dispatch_runtime.as_ref()
249    }
250}
251
252// ---------------------------------------------------------------------------
253// Request
254// ---------------------------------------------------------------------------
255
256/// State of the request body.
257enum RequestBody {
258    /// Still streaming from the connection.
259    Pending(Option<Incoming>),
260    /// Fully buffered, replayable.
261    Buffered(Vec<u8>),
262    /// Parsed as a multipart form.
263    Form(Arc<MultipartForm>),
264}
265
266/// Concrete implementation of IHttpRequest.
267pub struct HttpRequest {
268    method: String,
269    path: String,
270    headers: HashMap<String, String>,
271    query_params: HashMap<String, String>,
272    query_multi: HashMap<String, Vec<String>>,
273    route_params: HashMap<String, String>,
274    route_pattern: Option<String>,
275    remote_addr: Option<SocketAddr>,
276    body: Mutex<RequestBody>,
277    limits: BodyLimits,
278}
279
280#[async_trait::async_trait]
281impl IHttpRequest for HttpRequest {
282    fn method(&self) -> &str {
283        &self.method
284    }
285
286    fn path(&self) -> &str {
287        &self.path
288    }
289
290    /// Header values are looked up case-insensitively, as HTTP requires.
291    fn header(&self, name: &str) -> Option<&str> {
292        if let Some(value) = self.headers.get(name) {
293            return Some(value.as_str());
294        }
295        let lower = name.to_ascii_lowercase();
296        self.headers.get(&lower).map(|s| s.as_str())
297    }
298
299    fn query(&self) -> &HashMap<String, String> {
300        &self.query_params
301    }
302
303    fn query_all(&self) -> Option<&HashMap<String, Vec<String>>> {
304        Some(&self.query_multi)
305    }
306
307    fn remote_addr(&self) -> Option<SocketAddr> {
308        self.remote_addr
309    }
310
311    fn route_params(&self) -> &HashMap<String, String> {
312        &self.route_params
313    }
314
315    fn route_params_mut(&mut self) -> &mut HashMap<String, String> {
316        &mut self.route_params
317    }
318
319    fn route_pattern(&self) -> Option<&str> {
320        self.route_pattern.as_deref()
321    }
322
323    fn route_pattern_mut(&mut self) -> &mut Option<String> {
324        &mut self.route_pattern
325    }
326
327    async fn body_bytes(&mut self) -> Result<Vec<u8>> {
328        let mut guard = self.body.lock().await;
329        let incoming = match &mut *guard {
330            RequestBody::Buffered(bytes) => return Ok(bytes.clone()),
331            RequestBody::Form(_) => {
332                return Err(Error::State(
333                    "request body was already read as multipart/form-data".to_string(),
334                ))
335            }
336            RequestBody::Pending(slot) => slot.take(),
337        };
338
339        let Some(incoming) = incoming else {
340            return Err(Error::State(
341                "request body has already been consumed".to_string(),
342            ));
343        };
344
345        let bytes = read_body(incoming, self.limits.body).await?;
346        *guard = RequestBody::Buffered(bytes.clone());
347        Ok(bytes)
348    }
349
350    async fn body_text(&mut self) -> Result<String> {
351        let bytes = self.body_bytes().await?;
352        String::from_utf8(bytes)
353            .map_err(|_| Error::Validation("request body is not valid UTF-8".to_string()))
354    }
355
356    async fn multipart(&mut self) -> Result<Arc<MultipartForm>> {
357        let mut guard = self.body.lock().await;
358
359        match &*guard {
360            RequestBody::Form(form) => return Ok(Arc::clone(form)),
361            RequestBody::Buffered(_) => {
362                return Err(Error::UnsupportedMediaType(
363                    "request body was already read as a non-multipart body".to_string(),
364                ))
365            }
366            RequestBody::Pending(_) => {}
367        }
368
369        if !self.is_multipart() {
370            return Err(Error::UnsupportedMediaType(
371                "expected Content-Type: multipart/form-data".to_string(),
372            ));
373        }
374
375        let raw_content_type = self.header("content-type").unwrap_or("");
376        let boundary = multipart_boundary(raw_content_type).ok_or_else(|| {
377            Error::Validation(
378                "multipart/form-data request is missing its boundary parameter".to_string(),
379            )
380        })?;
381
382        let incoming = match &mut *guard {
383            RequestBody::Pending(slot) => slot.take(),
384            _ => None,
385        };
386
387        let Some(body) = incoming else {
388            return Err(Error::State(
389                "request body has already been consumed".to_string(),
390            ));
391        };
392
393        let form = Arc::new(parse_multipart(body, &boundary, &self.limits).await?);
394        *guard = RequestBody::Form(Arc::clone(&form));
395        Ok(form)
396    }
397}
398
399/// Drain a body stream into memory, enforcing `limit`.
400async fn read_body(incoming: Incoming, limit: usize) -> Result<Vec<u8>> {
401    let mut out = Vec::new();
402    let mut total = 0usize;
403    let mut stream = incoming.into_data_stream();
404
405    while let Some(chunk) = stream.next().await {
406        let chunk = chunk.map_err(|err| {
407            // Transport failure while reading the body: the client went away or
408            // the framing is broken. Log the details, never serve them.
409            tracing::debug!(error = %err, "failed to read request body");
410            Error::State("failed to read request body".to_string())
411        })?;
412        total = total.saturating_add(chunk.len());
413        if total > limit {
414            return Err(Error::PayloadTooLarge(format!(
415                "request body exceeds the {limit}-byte limit"
416            )));
417        }
418        out.extend_from_slice(&chunk);
419    }
420
421    Ok(out)
422}
423
424/// Whether a raw `Content-Type` header value declares a multipart form.
425fn is_multipart_content_type(content_type: Option<&String>) -> bool {
426    content_type
427        .map(|value| {
428            value
429                .split(';')
430                .next()
431                .unwrap_or("")
432                .trim()
433                .eq_ignore_ascii_case("multipart/form-data")
434        })
435        .unwrap_or(false)
436}
437
438/// Extract the `boundary` parameter of a `multipart/form-data` content type.
439fn multipart_boundary(content_type: &str) -> Option<String> {
440    for part in content_type.split(';').skip(1) {
441        let part = part.trim();
442        let Some((key, value)) = part.split_once('=') else {
443            continue;
444        };
445        if key.trim().eq_ignore_ascii_case("boundary") {
446            let value = value.trim().trim_matches('"');
447            if !value.is_empty() {
448                return Some(value.to_string());
449            }
450        }
451    }
452    None
453}
454
455/// Stream a multipart body into a [`MultipartForm`].
456///
457/// Text fields are buffered (bounded by `limits.field`); file parts stream into
458/// a [`FormFileBuilder`], which keeps small files in memory and spools large
459/// ones to disk. Nothing is copied twice, so peak memory stays bounded no
460/// matter how large the upload is.
461async fn parse_multipart(
462    incoming: Incoming,
463    boundary: &str,
464    limits: &BodyLimits,
465) -> Result<MultipartForm> {
466    let stream = incoming.into_data_stream();
467    let mut multipart = multer::Multipart::new(stream, boundary);
468    let mut form = MultipartForm::new();
469    let mut total: u64 = 0;
470
471    while let Some(field) = multipart.next_field().await.map_err(multer_error)? {
472        let field_name = field.name().unwrap_or_default().to_string();
473        if field_name.is_empty() {
474            continue;
475        }
476        let file_name = field.file_name().map(str::to_string);
477        let content_type = field.content_type().map(|m| m.to_string());
478
479        match file_name {
480            Some(file_name) => {
481                let mut builder = FormFileBuilder::new(field_name.clone(), file_name, content_type)
482                    .memory_threshold(limits.memory_threshold)
483                    .max_size(limits.file)
484                    .spool_dir(limits.spool_dir.clone());
485
486                let mut field = field;
487                while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
488                    total = charge(&mut total, chunk.len(), limits.multipart_body)?;
489                    builder.write(&chunk).await?;
490                }
491                form.push_file(builder.finish().await?);
492            }
493            None => {
494                let mut value: Vec<u8> = Vec::new();
495                let mut field = field;
496                while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
497                    total = charge(&mut total, chunk.len(), limits.multipart_body)?;
498                    if value.len() + chunk.len() > limits.field {
499                        return Err(Error::PayloadTooLarge(format!(
500                            "form field `{field_name}` exceeds the {}-byte limit",
501                            limits.field
502                        )));
503                    }
504                    value.extend_from_slice(&chunk);
505                }
506                let text = String::from_utf8(value).map_err(|_| {
507                    Error::Validation(format!("form field `{field_name}` is not valid UTF-8"))
508                })?;
509                form.push_field(field_name, text);
510            }
511        }
512    }
513
514    Ok(form)
515}
516
517/// Add `len` to the running total, failing when the request limit is exceeded.
518fn charge(total: &mut u64, len: usize, limit: u64) -> Result<u64> {
519    *total = total.saturating_add(len as u64);
520    if *total > limit {
521        return Err(Error::PayloadTooLarge(format!(
522            "multipart request exceeds the {limit}-byte limit"
523        )));
524    }
525    Ok(*total)
526}
527
528fn multer_error(err: multer::Error) -> Error {
529    match err {
530        multer::Error::FieldSizeExceeded { limit, .. } => {
531            Error::PayloadTooLarge(format!("multipart field exceeds the {limit}-byte limit"))
532        }
533        multer::Error::StreamSizeExceeded { limit } => {
534            Error::PayloadTooLarge(format!("multipart request exceeds the {limit}-byte limit"))
535        }
536        multer::Error::IncompleteFieldData { .. } | multer::Error::IncompleteStream => {
537            Error::Validation("multipart body ended unexpectedly".to_string())
538        }
539        other => Error::Validation(format!("invalid multipart body: {other}")),
540    }
541}
542
543// ---------------------------------------------------------------------------
544// Response
545// ---------------------------------------------------------------------------
546
547/// Concrete implementation of IHttpResponse.
548pub struct HttpResponse {
549    status: u16,
550    headers: HashMap<String, String>,
551    body: Option<ResponseBody>,
552}
553
554impl HttpResponse {
555    pub fn new(status: u16) -> Self {
556        Self {
557            status,
558            headers: HashMap::new(),
559            body: None,
560        }
561    }
562}
563
564#[async_trait::async_trait]
565impl IHttpResponse for HttpResponse {
566    fn status(&self) -> u16 {
567        self.status
568    }
569
570    fn has_body(&self) -> bool {
571        self.body.is_some()
572    }
573
574    fn set_status(&mut self, code: u16) {
575        self.status = code;
576    }
577
578    fn set_header(&mut self, key: &str, value: &str) {
579        self.headers.insert(key.to_string(), value.to_string());
580    }
581
582    fn remove_header(&mut self, key: &str) {
583        self.headers.remove(key);
584    }
585
586    async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
587        self.body = Some(ResponseBody::Bytes(data));
588        Ok(())
589    }
590
591    async fn write_text(&mut self, text: &str) -> Result<()> {
592        self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
593        Ok(())
594    }
595
596    async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
597        self.body = Some(body);
598        Ok(())
599    }
600
601    fn body_bytes(&self) -> Vec<u8> {
602        self.body
603            .as_ref()
604            .and_then(ResponseBody::as_bytes)
605            .map(<[u8]>::to_vec)
606            .unwrap_or_default()
607    }
608
609    fn header(&self, key: &str) -> Option<&str> {
610        self.headers.get(key).map(|s| s.as_str())
611    }
612}
613
614// ---------------------------------------------------------------------------
615// Response finalisation
616// ---------------------------------------------------------------------------
617
618async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
619    let HttpResponse {
620        status,
621        mut headers,
622        body,
623    } = resp;
624
625    match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
626        ResponseBody::Bytes(bytes) => {
627            default_cache_control(&mut headers, false);
628            bytes_response(status, headers, bytes)
629        }
630        ResponseBody::File(file) => {
631            default_cache_control(&mut headers, true);
632            file_response(req, status, headers, file).await
633        }
634    }
635}
636
637/// Apply the framework's default `Cache-Control` when the response has none.
638///
639/// Ordinary API responses must not be stored; a file response with validators is
640/// exactly the case where `no-store` would defeat the `ETag` we just sent, so it
641/// revalidates instead. Anything the application set explicitly wins.
642fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
643    if headers.contains_key("cache-control") {
644        return;
645    }
646    headers.insert(
647        "cache-control".to_string(),
648        if is_file {
649            "public, max-age=0, must-revalidate".to_string()
650        } else {
651            "no-store".to_string()
652        },
653    );
654}
655
656/// A file body whose source has been resolved to something streamable.
657enum PreparedSource {
658    /// An open file handle; seeks to serve ranges.
659    Path(tokio::fs::File),
660    /// A reader that can seek, so ranges are served by seeking.
661    Seekable(Box<dyn SeekableReader>),
662    /// A reader of unknown length; chunked, no ranges.
663    Stream(Box<dyn AsyncRead + Send + Unpin>),
664}
665
666impl PreparedSource {
667    fn is_seekable(&self) -> bool {
668        matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
669    }
670}
671
672/// Serve a file response with validators, conditional handling and byte ranges.
673///
674/// Works the same way for a path, a seekable stream and a plain stream; the only
675/// differences are which HTTP features the source can support.
676async fn file_response(
677    req: &HttpRequest,
678    status: u16,
679    mut headers: HashMap<String, String>,
680    file: FileBody,
681) -> hyper::Response<RespBody> {
682    let FileBody {
683        source,
684        content_type,
685        file_name,
686        disposition,
687        enable_range_processing,
688        last_modified,
689        entity_tag,
690    } = file;
691
692    // ── Resolve the source, its length and its validators ──
693    let mut modified = last_modified;
694    let mut etag = entity_tag;
695    let mut source_path: Option<PathBuf> = None;
696    let known_len;
697
698    let prepared = match source {
699        FileSource::Path(path) => {
700            // Open once and take length/mtime from the same handle: one syscall
701            // pair instead of `metadata()` followed by `open()`.
702            let handle = match tokio::fs::File::open(&path).await {
703                Ok(handle) => handle,
704                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
705                    return problem_bytes(404, headers, "Not Found")
706                }
707                Err(err) => {
708                    tracing::error!(
709                        path = %path.display(),
710                        error = %err,
711                        "cannot open file for download"
712                    );
713                    return problem_bytes(500, headers, "an unexpected error occurred");
714                }
715            };
716            let metadata = match handle.metadata().await {
717                Ok(metadata) => metadata,
718                Err(err) => {
719                    tracing::error!(
720                        path = %path.display(),
721                        error = %err,
722                        "cannot stat file for download"
723                    );
724                    return problem_bytes(500, headers, "an unexpected error occurred");
725                }
726            };
727            if !metadata.is_file() {
728                return problem_bytes(404, headers, "Not Found");
729            }
730
731            let len = metadata.len();
732            if modified.is_none() {
733                modified = metadata.modified().ok();
734            }
735            if etag.is_none() {
736                etag = Some(etag_for(len, modified));
737            }
738            known_len = Some(len);
739            source_path = Some(path);
740            PreparedSource::Path(handle)
741        }
742        FileSource::SeekableStream { reader, len } => {
743            known_len = Some(len);
744            PreparedSource::Seekable(reader)
745        }
746        FileSource::Stream(reader) => {
747            known_len = None;
748            PreparedSource::Stream(reader)
749        }
750    };
751
752    // RFC 7232 §2.2.2: an origin server must not send a `Last-Modified` later
753    // than the response generation time, and must not treat such a date as a
754    // precondition. Restored archives and clock skew produce future mtimes, and
755    // they would otherwise make `If-Modified-Since` and date-based `If-Range`
756    // answer nonsense. The ETag above was already derived from the real mtime,
757    // so content changes are still detected.
758    let modified = modified.filter(|value| *value <= SystemTime::now());
759
760    // ── Representation metadata ──
761    if !headers.contains_key("content-type") {
762        let resolved = content_type
763            .or_else(|| source_path.as_deref().and_then(guess_content_type))
764            .unwrap_or_else(|| "application/octet-stream".to_string());
765        headers.insert("content-type".to_string(), resolved);
766    }
767    if !headers.contains_key("content-disposition") {
768        if let Some(name) = &file_name {
769            headers.insert(
770                "content-disposition".to_string(),
771                content_disposition_value(disposition, name),
772            );
773        }
774    }
775    if let Some(tag) = &etag {
776        headers.insert("etag".to_string(), tag.clone());
777    }
778    if let Some(modified) = modified {
779        headers.insert(
780            "last-modified".to_string(),
781            httpdate::fmt_http_date(modified),
782        );
783    }
784
785    // `Accept-Ranges: none` tells download managers not to bother trying, which
786    // is more useful than the header's absence (absence means "unknown").
787    let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
788    headers.insert(
789        "accept-ranges".to_string(),
790        if range_capable { "bytes" } else { "none" }.to_string(),
791    );
792
793    // ── Conditional GET: the representation is unchanged ──
794    if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
795        headers.remove("content-length");
796        return bytes_response(304, headers, Vec::new());
797    }
798
799    // ── Byte ranges ──
800    let mut ranges: Vec<ByteRange> = Vec::new();
801    if status == 200 && range_capable {
802        let len = known_len.expect("range-capable sources know their length");
803        if let Some(raw) = req.header("range") {
804            // `If-Range` gates the whole range request: if the client's copy is
805            // stale, the correct answer is the *full* representation, not a
806            // range from a different version.
807            if if_range_matches(req, etag.as_deref(), modified) {
808                match parse_ranges(raw, len) {
809                    RangeOutcome::Ignored => {}
810                    RangeOutcome::Unsatisfiable => {
811                        headers.insert("content-range".to_string(), format!("bytes */{len}"));
812                        return problem_bytes(416, headers, "Range Not Satisfiable");
813                    }
814                    RangeOutcome::Satisfiable(parsed) => ranges = parsed,
815                }
816            }
817        }
818    }
819
820    // ── Body ──
821    //
822    // A multi-range request needs to re-read the same source from several
823    // offsets. Only a file handle can do that; a range-capable *stream* cannot
824    // be restarted, so those requests are answered with the full representation
825    // (RFC 7233 §3.1 lets a server ignore `Range`).
826    if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
827        ranges.clear();
828    }
829    if ranges.len() > 1 {
830        return multipart_byteranges_response(prepared, headers, ranges, known_len);
831    }
832
833    let (status, start, count) = match ranges.first() {
834        Some(range) => {
835            let len = known_len.expect("range-capable sources know their length");
836            headers.insert(
837                "content-range".to_string(),
838                format!("bytes {}-{}/{len}", range.start, range.end),
839            );
840            (206, range.start, range.len())
841        }
842        None => (status, 0, known_len.unwrap_or(0)),
843    };
844
845    match prepared {
846        PreparedSource::Path(mut handle) => {
847            headers.insert("content-length".to_string(), count.to_string());
848            if count == 0 {
849                return bytes_response(status, headers, Vec::new());
850            }
851            if start > 0 {
852                if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
853                    tracing::error!(error = %err, "cannot seek file for ranged response");
854                    return problem_bytes(500, headers, "an unexpected error occurred");
855                }
856            }
857            build_response(status, headers, reader_body(handle.take(count)))
858        }
859        PreparedSource::Seekable(mut reader) => {
860            headers.insert("content-length".to_string(), count.to_string());
861            if count == 0 {
862                return bytes_response(status, headers, Vec::new());
863            }
864            if start > 0 {
865                if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
866                    tracing::error!(error = %err, "cannot seek stream for ranged response");
867                    return problem_bytes(500, headers, "an unexpected error occurred");
868                }
869            }
870            build_response(status, headers, reader_body(reader.take(count)))
871        }
872        PreparedSource::Stream(reader) => {
873            // Unknown length: chunked transfer encoding, no `Content-Length`.
874            // Nothing is buffered.
875            build_response(status, headers, reader_body(reader))
876        }
877    }
878}
879
880/// The media type of the parts inside a `multipart/byteranges` response.
881///
882/// The response-level `Content-Type` is about to be replaced, so the original
883/// representation type has to be captured first.
884fn representation_content_type(headers: &HashMap<String, String>) -> String {
885    headers
886        .get("content-type")
887        .filter(|value| !value.starts_with("multipart/byteranges"))
888        .cloned()
889        .unwrap_or_else(|| "application/octet-stream".to_string())
890}
891
892/// Serve several disjoint ranges as one `multipart/byteranges` response.
893///
894/// Ranges are coalesced before this point, so the parts are disjoint and the
895/// total payload never exceeds the file size — no traffic amplification. The
896/// response length is computed exactly rather than falling back to chunked,
897/// because every part's size is known up front.
898fn multipart_byteranges_response(
899    prepared: PreparedSource,
900    mut headers: HashMap<String, String>,
901    ranges: Vec<ByteRange>,
902    known_len: Option<u64>,
903) -> hyper::Response<RespBody> {
904    let PreparedSource::Path(handle) = prepared else {
905        unreachable!("multi-range is only routed here for path sources");
906    };
907
908    let total_len = known_len.unwrap_or(0);
909    let media_type = representation_content_type(&headers);
910    let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
911
912    // Pre-render each part's header block so the total length is exact by
913    // construction and the stream only has to emit bytes.
914    let parts: Vec<(Bytes, ByteRange)> = ranges
915        .iter()
916        .map(|range| {
917            let header = format!(
918                "--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
919                range.start, range.end
920            );
921            (Bytes::from(header), *range)
922        })
923        .collect();
924    let closing = Bytes::from(format!("--{boundary}--\r\n"));
925
926    let content_length: u64 = parts
927        .iter()
928        .map(|(header, range)| header.len() as u64 + range.len() + 2)
929        .sum::<u64>()
930        + closing.len() as u64;
931
932    headers.insert(
933        "content-type".to_string(),
934        format!("multipart/byteranges; boundary={boundary}"),
935    );
936    headers.insert("content-length".to_string(), content_length.to_string());
937    // `Content-Range` is per part here, so the message must not carry one.
938    headers.remove("content-range");
939
940    let stream = multipart_stream(handle, parts, closing);
941    build_response(
942        206,
943        headers,
944        BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
945    )
946}
947
948/// Chunk size for each `multipart/byteranges` part read.
949const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
950
951/// State carried between polls of the `multipart/byteranges` stream.
952struct MultipartState {
953    handle: tokio::fs::File,
954    parts: std::vec::IntoIter<(Bytes, ByteRange)>,
955    closing: Bytes,
956    /// Header bytes still to emit for the current part.
957    pending_header: Bytes,
958    phase: MultipartPhase,
959    buffer: bytes::BytesMut,
960}
961
962enum MultipartPhase {
963    /// Move to the next part's header, or close the body.
964    NextPart,
965    /// Stream the current part's payload.
966    Payload(u64),
967    /// Emit the `\r\n` that terminates a part.
968    Terminator,
969    /// Emit the closing delimiter.
970    Closing,
971    Done,
972}
973
974/// Build the `multipart/byteranges` body as a stream of frames.
975///
976/// Written as a `Stream` rather than a hand-rolled `AsyncRead`: the byte layout
977/// is a sequence of independent chunks, so `unfold` expresses it directly and
978/// there is no `poll_*` plumbing to get wrong.
979fn multipart_stream(
980    handle: tokio::fs::File,
981    parts: Vec<(Bytes, ByteRange)>,
982    closing: Bytes,
983) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
984    let state = MultipartState {
985        handle,
986        parts: parts.into_iter(),
987        closing,
988        pending_header: Bytes::new(),
989        phase: MultipartPhase::NextPart,
990        buffer: bytes::BytesMut::new(),
991    };
992
993    futures_util::stream::unfold(state, |mut state| async move {
994        loop {
995            match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
996                MultipartPhase::NextPart => match state.parts.next() {
997                    Some((header, range)) => {
998                        if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
999                            return Some((Err(err), state));
1000                        }
1001                        state.pending_header = header;
1002                        state.phase = MultipartPhase::Payload(range.len());
1003                    }
1004                    None => state.phase = MultipartPhase::Closing,
1005                },
1006                MultipartPhase::Payload(remaining) => {
1007                    if remaining == 0 {
1008                        state.phase = MultipartPhase::Terminator;
1009                        continue;
1010                    }
1011
1012                    // Read at most one chunk, never past the part boundary.
1013                    let limit = MULTIPART_CHUNK.min(remaining as usize);
1014                    state.buffer.clear();
1015                    state.buffer.reserve(limit);
1016                    let read = match state.handle.read_buf(&mut state.buffer).await {
1017                        Ok(read) => read,
1018                        Err(err) => return Some((Err(err), state)),
1019                    };
1020                    if read == 0 {
1021                        return Some((
1022                            Err(std::io::Error::new(
1023                                std::io::ErrorKind::UnexpectedEof,
1024                                "file changed size while serving byte ranges",
1025                            )),
1026                            state,
1027                        ));
1028                    }
1029                    let keep = read.min(limit);
1030                    state.buffer.truncate(keep);
1031                    let chunk = state.buffer.split().freeze();
1032
1033                    // The part header is emitted immediately before its first
1034                    // payload byte, so a part is one contiguous run of frames.
1035                    let header = std::mem::take(&mut state.pending_header);
1036                    let frame = if header.is_empty() {
1037                        chunk
1038                    } else {
1039                        let mut combined =
1040                            bytes::BytesMut::with_capacity(header.len() + chunk.len());
1041                        combined.extend_from_slice(&header);
1042                        combined.extend_from_slice(&chunk);
1043                        combined.freeze()
1044                    };
1045
1046                    state.phase = MultipartPhase::Payload(remaining - keep as u64);
1047                    return Some((Ok(Frame::data(frame)), state));
1048                }
1049                MultipartPhase::Terminator => {
1050                    state.phase = MultipartPhase::NextPart;
1051                    return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
1052                }
1053                MultipartPhase::Closing => {
1054                    state.phase = MultipartPhase::Done;
1055                    let closing = std::mem::take(&mut state.closing);
1056                    return Some((Ok(Frame::data(closing)), state));
1057                }
1058                MultipartPhase::Done => return None,
1059            }
1060        }
1061    })
1062}
1063
1064/// Read buffer used for every streamed response body.
1065///
1066/// Shared with the upload path so both sides of a transfer use the same size.
1067const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
1068
1069/// Turn any async reader into a streaming response body.
1070fn reader_body<R>(reader: R) -> RespBody
1071where
1072    R: tokio::io::AsyncRead + Send + 'static,
1073{
1074    let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
1075    let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
1076    BodyExt::boxed_unsync(body)
1077}
1078
1079fn bytes_response(
1080    status: u16,
1081    headers: HashMap<String, String>,
1082    bytes: Vec<u8>,
1083) -> hyper::Response<RespBody> {
1084    build_response(status, headers, full_body(bytes))
1085}
1086
1087fn full_body(bytes: Vec<u8>) -> RespBody {
1088    Full::new(Bytes::from(bytes))
1089        .map_err(|never: std::convert::Infallible| match never {})
1090        .boxed_unsync()
1091}
1092
1093/// Build an RFC 7807 `application/problem+json` response for a file error.
1094fn problem_bytes(
1095    status: u16,
1096    mut headers: HashMap<String, String>,
1097    detail: &str,
1098) -> hyper::Response<RespBody> {
1099    headers.insert(
1100        "content-type".to_string(),
1101        "application/problem+json".to_string(),
1102    );
1103    bytes_response(
1104        status,
1105        headers,
1106        problem_to_bytes(&build_problem(status, detail)),
1107    )
1108}
1109
1110fn build_response(
1111    status: u16,
1112    headers: HashMap<String, String>,
1113    body: RespBody,
1114) -> hyper::Response<RespBody> {
1115    let mut builder = hyper::Response::builder().status(status);
1116    for (key, value) in &headers {
1117        builder = builder.header(key.as_str(), value.as_str());
1118    }
1119    builder.body(body).unwrap_or_else(|_| {
1120        hyper::Response::builder()
1121            .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
1122            .body(full_body(b"Internal Server Error".to_vec()))
1123            .expect("static error response is valid")
1124    })
1125}
1126
1127/// Result of parsing a `Range` header.
1128#[derive(Debug, PartialEq, Eq)]
1129enum RangeOutcome {
1130    /// Syntactically invalid: the whole header must be ignored (RFC 7233 §2.1).
1131    Ignored,
1132    /// Valid but nothing overlaps the representation: answer `416`.
1133    Unsatisfiable,
1134    /// One or more disjoint, coalesced ranges, in ascending order.
1135    Satisfiable(Vec<ByteRange>),
1136}
1137
1138/// Maximum number of parts to answer in one `multipart/byteranges` response.
1139///
1140/// Ranges are coalesced first, so the parts are disjoint and the total payload
1141/// can never exceed the file size; this cap only bounds the *framing* overhead
1142/// and the per-part seeks. A client asking for more gets the full
1143/// representation, which RFC 7233 §3.1 permits.
1144const MAX_RANGE_PARTS: usize = 8;
1145
1146/// Parse a byte-range-set (RFC 7233 §2.1) and coalesce it.
1147///
1148/// Coalescing matters for more than tidiness: without it a client could ask for
1149/// the same range eight times and make the server send the file eight times
1150/// over. Merging overlapping and adjacent ranges guarantees the parts are
1151/// disjoint, so the response can never be larger than the file.
1152fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
1153    let value = value.trim();
1154    let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
1155        return RangeOutcome::Ignored;
1156    };
1157
1158    let mut ranges: Vec<ByteRange> = Vec::new();
1159    let mut saw_valid_spec = false;
1160
1161    for part in spec.split(',') {
1162        let part = part.trim();
1163        if part.is_empty() {
1164            continue;
1165        }
1166        match parse_single_range(part, len) {
1167            // One malformed member invalidates the whole header.
1168            SingleRange::Malformed => return RangeOutcome::Ignored,
1169            SingleRange::Unsatisfiable => saw_valid_spec = true,
1170            SingleRange::Satisfiable(range) => {
1171                saw_valid_spec = true;
1172                ranges.push(range);
1173            }
1174        }
1175    }
1176
1177    if ranges.is_empty() {
1178        return if saw_valid_spec {
1179            RangeOutcome::Unsatisfiable
1180        } else {
1181            RangeOutcome::Ignored
1182        };
1183    }
1184
1185    ranges.sort_by_key(|range| range.start);
1186    let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
1187    for range in ranges {
1188        match merged.last_mut() {
1189            // `+ 1` merges adjacent ranges too (`0-9,10-19` is one read).
1190            Some(last) if range.start <= last.end.saturating_add(1) => {
1191                last.end = last.end.max(range.end);
1192            }
1193            _ => merged.push(range),
1194        }
1195    }
1196
1197    if merged.len() > MAX_RANGE_PARTS {
1198        return RangeOutcome::Ignored;
1199    }
1200
1201    RangeOutcome::Satisfiable(merged)
1202}
1203
1204enum SingleRange {
1205    /// Not a valid `byte-range-spec`.
1206    Malformed,
1207    /// Valid syntax, but outside the representation.
1208    Unsatisfiable,
1209    Satisfiable(ByteRange),
1210}
1211
1212/// Parse one `first-byte-pos "-" [last-byte-pos]` or `"-" suffix-length`.
1213fn parse_single_range(spec: &str, len: u64) -> SingleRange {
1214    let Some((start_text, end_text)) = spec.split_once('-') else {
1215        return SingleRange::Malformed;
1216    };
1217    let (start_text, end_text) = (start_text.trim(), end_text.trim());
1218
1219    if start_text.is_empty() {
1220        // Suffix range: the last N bytes.
1221        let Ok(suffix) = end_text.parse::<u64>() else {
1222            return SingleRange::Malformed;
1223        };
1224        if suffix == 0 || len == 0 {
1225            return SingleRange::Unsatisfiable;
1226        }
1227        return SingleRange::Satisfiable(ByteRange {
1228            start: len.saturating_sub(suffix),
1229            end: len - 1,
1230        });
1231    }
1232
1233    let Ok(start) = start_text.parse::<u64>() else {
1234        return SingleRange::Malformed;
1235    };
1236    if start >= len {
1237        return SingleRange::Unsatisfiable;
1238    }
1239
1240    let end = if end_text.is_empty() {
1241        len - 1
1242    } else {
1243        match end_text.parse::<u64>() {
1244            Ok(end) => end.min(len - 1),
1245            Err(_) => return SingleRange::Malformed,
1246        }
1247    };
1248
1249    if end < start {
1250        return SingleRange::Unsatisfiable;
1251    }
1252
1253    SingleRange::Satisfiable(ByteRange { start, end })
1254}
1255
1256fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
1257    if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
1258        Some(&value[prefix.len()..])
1259    } else {
1260        None
1261    }
1262}
1263
1264/// A weak validator derived from size and modification time.
1265///
1266/// This is the same scheme nginx and most static servers use: free to compute,
1267/// and it changes whenever the file does. Full `SystemTime` precision is used so
1268/// an in-place edit that keeps the size is still detected within the same
1269/// second — `Last-Modified` cannot express that, but an `ETag` can.
1270///
1271/// It is emitted as a **strong** tag because download resumption needs one:
1272/// RFC 7233 §3.2 requires a strong validator for `If-Range`, so a weak tag would
1273/// silently disable resumable downloads. Callers that need a byte-exact
1274/// validator should set one explicitly with `FileBody::entity_tag`.
1275fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
1276    match modified {
1277        Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
1278        None => format!("\"{len:x}\""),
1279    }
1280}
1281
1282fn epoch_secs(time: SystemTime) -> u64 {
1283    time.duration_since(UNIX_EPOCH)
1284        .map(|d| d.as_secs())
1285        .unwrap_or(0)
1286}
1287
1288fn epoch_nanos(time: SystemTime) -> u128 {
1289    time.duration_since(UNIX_EPOCH)
1290        .map(|d| d.as_nanos())
1291        .unwrap_or(0)
1292}
1293
1294/// Whether an entity-tag is weak (`W/"..."`).
1295fn is_weak_etag(tag: &str) -> bool {
1296    tag.trim_start().starts_with("W/")
1297}
1298
1299/// Strip the weakness marker so two tags can be compared by value.
1300fn etag_value(tag: &str) -> &str {
1301    tag.trim().trim_start_matches("W/")
1302}
1303
1304/// Whether the client's cached representation is still current.
1305///
1306/// Uses **weak** comparison, per RFC 7232 §3.2: `If-None-Match` only asks
1307/// whether the content is semantically the same, so a weak tag is enough. A
1308/// response without validators can never answer `304`.
1309fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1310    if let Some(if_none_match) = req.header("if-none-match") {
1311        let Some(etag) = etag else {
1312            return false;
1313        };
1314        let expected = etag_value(etag);
1315        return if_none_match.split(',').any(|candidate| {
1316            let candidate = candidate.trim();
1317            candidate == "*" || etag_value(candidate) == expected
1318        });
1319    }
1320
1321    if let Some(if_modified_since) = req.header("if-modified-since") {
1322        if let Some(modified) = modified {
1323            if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
1324                return epoch_secs(modified) <= epoch_secs(since);
1325            }
1326        }
1327    }
1328
1329    false
1330}
1331
1332/// Whether a `Range` header should be honoured, per RFC 7233 §3.2.
1333///
1334/// `If-Range` is the precondition that makes resumption safe, so it is strict:
1335///
1336/// * an entity-tag must match by **strong** comparison — both tags must be
1337///   strong and byte-equal. A weak tag (ours or the client's) can never match,
1338///   because a weak validator does not promise byte-for-byte identity;
1339/// * a date must be at least as new as `Last-Modified`.
1340///
1341/// Returning `false` means "ignore `Range` and send the full representation".
1342fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1343    let Some(value) = req.header("if-range") else {
1344        return true;
1345    };
1346    let value = value.trim();
1347
1348    if value.starts_with('"') || value.starts_with("W/") {
1349        let Some(etag) = etag else {
1350            return false;
1351        };
1352        if is_weak_etag(value) || is_weak_etag(etag) {
1353            return false;
1354        }
1355        return value == etag.trim();
1356    }
1357
1358    match (httpdate::parse_http_date(value), modified) {
1359        (Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
1360        _ => false,
1361    }
1362}
1363
1364fn guess_content_type(path: &Path) -> Option<String> {
1365    mime_guess::from_path(path)
1366        .first()
1367        .map(|mime| mime.essence_str().to_string())
1368}
1369
1370// ---------------------------------------------------------------------------
1371// Query string helpers
1372// ---------------------------------------------------------------------------
1373
1374/// Parse a query string like "key1=val1&key2=val2" into both views.
1375///
1376/// Returns the last value per key (the compatibility view used by
1377/// `IHttpRequest::query`) and every value per key (used for `Vec<T>` fields).
1378/// Repeated keys keep their order; empty pairs are ignored.
1379fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
1380    let mut flat = HashMap::new();
1381    let mut multi: HashMap<String, Vec<String>> = HashMap::new();
1382    for pair in query.split('&') {
1383        if pair.is_empty() {
1384            continue;
1385        }
1386        let mut parts = pair.splitn(2, '=');
1387        if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
1388            let key = percent_decode(key);
1389            let value = percent_decode(value);
1390            flat.insert(key.clone(), value.clone());
1391            multi.entry(key).or_default().push(value);
1392        }
1393    }
1394    (flat, multi)
1395}
1396
1397/// Decode `%XX` escapes and `+` into a UTF-8 string.
1398///
1399/// Escapes are decoded as bytes and then validated as UTF-8, so multi-byte
1400/// characters (`%E4%B8%AD` → `中`) survive; the previous byte-per-char loop
1401/// mangled anything outside ASCII.
1402fn percent_decode(s: &str) -> String {
1403    if !s.contains('%') && !s.contains('+') {
1404        return s.to_string();
1405    }
1406
1407    let mut bytes = Vec::with_capacity(s.len());
1408    let mut chars = s.bytes();
1409    while let Some(byte) = chars.next() {
1410        match byte {
1411            b'%' => {
1412                let hi = chars.next().and_then(|c| (c as char).to_digit(16));
1413                let lo = chars.next().and_then(|c| (c as char).to_digit(16));
1414                match (hi, lo) {
1415                    (Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
1416                    _ => bytes.push(b'%'),
1417                }
1418            }
1419            b'+' => bytes.push(b' '),
1420            other => bytes.push(other),
1421        }
1422    }
1423    String::from_utf8_lossy(&bytes).into_owned()
1424}
1425
1426#[cfg(test)]
1427mod tests {
1428    use super::*;
1429
1430    #[test]
1431    fn parses_the_boundary_parameter() {
1432        assert_eq!(
1433            multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
1434            Some("----abc123")
1435        );
1436        assert_eq!(
1437            multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
1438                .as_deref(),
1439            Some("quoted")
1440        );
1441        assert_eq!(multipart_boundary("multipart/form-data"), None);
1442    }
1443
1444    fn ranges(start: u64, end: u64) -> RangeOutcome {
1445        RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
1446    }
1447
1448    #[test]
1449    fn parses_byte_ranges() {
1450        assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
1451        assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
1452        assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
1453        // Clamped to the end of the representation.
1454        assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
1455        // Case-insensitive unit, and whitespace around members is tolerated.
1456        assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
1457        assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
1458    }
1459
1460    #[test]
1461    fn rejects_unsatisfiable_and_malformed_ranges() {
1462        assert_eq!(
1463            parse_ranges("bytes=1000-", 1000),
1464            RangeOutcome::Unsatisfiable
1465        );
1466        assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
1467        assert_eq!(
1468            parse_ranges("bytes=500-100", 1000),
1469            RangeOutcome::Unsatisfiable
1470        );
1471        assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
1472        assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
1473        // RFC 7233 §2.1: one malformed member invalidates the whole header.
1474        assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
1475        assert_eq!(
1476            parse_ranges("bytes=0-9,abc-def", 1000),
1477            RangeOutcome::Ignored
1478        );
1479    }
1480
1481    #[test]
1482    fn suffix_range_on_empty_file_is_unsatisfiable() {
1483        assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
1484    }
1485
1486    #[test]
1487    fn multi_range_is_coalesced_into_disjoint_parts() {
1488        // Disjoint ranges stay separate, in ascending order.
1489        assert_eq!(
1490            parse_ranges("bytes=20-30,0-10", 1000),
1491            RangeOutcome::Satisfiable(vec![
1492                ByteRange { start: 0, end: 10 },
1493                ByteRange { start: 20, end: 30 },
1494            ])
1495        );
1496
1497        // Overlapping ranges merge, so a client cannot ask for the same bytes
1498        // twice and make the server send them twice.
1499        assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
1500
1501        // Adjacent ranges merge too — one read instead of two.
1502        assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
1503
1504        // Duplicates collapse completely: no traffic amplification.
1505        assert_eq!(
1506            parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
1507            ranges(0, 999)
1508        );
1509
1510        // An unsatisfiable member next to a satisfiable one is dropped, not fatal.
1511        assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
1512    }
1513
1514    #[test]
1515    fn too_many_parts_falls_back_to_the_full_representation() {
1516        let mut spec = String::from("bytes=");
1517        for index in 0..(MAX_RANGE_PARTS + 1) {
1518            if index > 0 {
1519                spec.push(',');
1520            }
1521            spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
1522        }
1523        assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
1524    }
1525
1526    #[test]
1527    fn etag_changes_with_size_and_mtime() {
1528        let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
1529        assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
1530        assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
1531        assert_ne!(
1532            etag_for(10, Some(base)),
1533            etag_for(10, Some(base + std::time::Duration::from_secs(1)))
1534        );
1535        // Sub-second precision matters: an in-place edit that keeps the size is
1536        // invisible to `Last-Modified`, but must change the ETag. (1 µs rather
1537        // than 1 ns, because Windows filesystem timestamps are 100 ns ticks.)
1538        assert_ne!(
1539            etag_for(10, Some(base)),
1540            etag_for(10, Some(base + std::time::Duration::from_micros(1)))
1541        );
1542    }
1543
1544    #[test]
1545    fn etag_strength_rules() {
1546        assert!(!is_weak_etag("\"abc\""));
1547        assert!(is_weak_etag("W/\"abc\""));
1548        assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
1549        assert_eq!(etag_value("\"abc\""), "\"abc\"");
1550    }
1551
1552    #[test]
1553    fn parses_query_strings() {
1554        let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
1555        assert_eq!(params.get("a").map(String::as_str), Some("1"));
1556        assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
1557        assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
1558        assert_eq!(
1559            multi.get("a").map(Vec::as_slice),
1560            Some(&["1".to_string()][..])
1561        );
1562    }
1563
1564    #[test]
1565    fn keeps_every_query_value_for_repeated_keys() {
1566        let (params, multi) = parse_query_string("id=1&id=2&id=3");
1567        assert_eq!(params.get("id").map(String::as_str), Some("3"));
1568        assert_eq!(
1569            multi.get("id").map(Vec::as_slice),
1570            Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
1571        );
1572    }
1573
1574    #[test]
1575    fn percent_decodes_utf8() {
1576        // Multi-byte characters must survive: the byte-per-char loop turned
1577        // `%E4%B8%AD` into mojibake.
1578        let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
1579        assert_eq!(params.get("q").map(String::as_str), Some("中文"));
1580        assert_eq!(params.get("x").map(String::as_str), Some("+"));
1581    }
1582}