sova_core/request/
mod.rs

1use crate::error::{Error, Result};
2use crate::response::{HttpBody, Response};
3use crate::server::collect_limited;
4use crate::state::{Extensions, StateMap};
5use bytes::Bytes;
6use http::{HeaderMap, HeaderName, HeaderValue, Method};
7use http_body_util::BodyExt;
8use rustc_hash::FxHashMap;
9use serde::de::DeserializeOwned;
10use std::str::FromStr;
11use std::sync::Arc;
12
13mod input;
14pub use input::{FormData, Upload, UploadRules};
15
16/// Request body: buffered bytes or a lazy stream (collected on demand).
17pub enum ReqBody {
18    Bytes(Bytes),
19    Stream(HttpBody),
20    /// Consumed by a prior body reader (`by` names the consumer).
21    Taken { by: &'static str },
22}
23
24/// Incoming HTTP request with Express-style helpers.
25pub struct Request {
26    pub method: Method,
27    pub path: String,
28    pub headers: HeaderMap,
29    pub params: FxHashMap<String, String>,
30    pub query: FxHashMap<String, String>,
31    /// Scheme (`http` / `https`), possibly from `X-Forwarded-Proto` when trust_proxy.
32    pub(crate) scheme: String,
33    /// Host (no port stripping beyond what the client sent).
34    pub(crate) host: String,
35    /// Raw query string without `?` (for `query_as`).
36    pub(crate) raw_query: String,
37    pub(crate) body: ReqBody,
38    pub(crate) body_limit: usize,
39    pub(crate) state: Arc<StateMap>,
40    pub(crate) extensions: Extensions,
41}
42
43/// Builder for test / embedded requests. `state` and `extensions` stay empty —
44/// [`crate::App::handle`] fills router state.
45pub struct RequestBuilder {
46    method: Method,
47    path: String,
48    headers: HeaderMap,
49    body: Bytes,
50    query: FxHashMap<String, String>,
51    raw_query: String,
52    scheme: String,
53    host: String,
54    body_limit: usize,
55}
56
57impl RequestBuilder {
58    pub fn method(mut self, method: Method) -> Self {
59        self.method = method;
60        self
61    }
62
63    pub fn path(mut self, path: impl Into<String>) -> Self {
64        let path = path.into();
65        match path.split_once('?') {
66            Some((p, q)) => {
67                self.path = p.to_string();
68                self.raw_query = q.to_string();
69                self.query = parse_query(q);
70            }
71            None => {
72                self.path = path;
73            }
74        }
75        self
76    }
77
78    pub fn header(mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
79        if let (Ok(name), Ok(value)) = (
80            HeaderName::from_bytes(name.as_ref().as_bytes()),
81            HeaderValue::from_str(value.as_ref()),
82        ) {
83            self.headers.insert(name, value);
84        }
85        self
86    }
87
88    pub fn body(mut self, body: impl Into<Bytes>) -> Self {
89        self.body = body.into();
90        self
91    }
92
93    pub fn body_limit(mut self, limit: usize) -> Self {
94        self.body_limit = limit;
95        self
96    }
97
98    pub fn query_param(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
99        self.query.insert(key.into(), value.into());
100        self.raw_query = serde_urlencoded::to_string(&self.query).unwrap_or_default();
101        self
102    }
103
104    pub fn scheme(mut self, scheme: impl Into<String>) -> Self {
105        self.scheme = scheme.into();
106        self
107    }
108
109    pub fn host(mut self, host: impl Into<String>) -> Self {
110        self.host = host.into();
111        self
112    }
113
114    pub fn build(self) -> Request {
115        Request {
116            method: self.method,
117            path: self.path,
118            headers: self.headers,
119            params: FxHashMap::default(),
120            query: self.query,
121            scheme: self.scheme,
122            host: self.host,
123            raw_query: self.raw_query,
124            body: ReqBody::Bytes(self.body),
125            body_limit: self.body_limit,
126            state: Arc::new(StateMap::new()),
127            extensions: Extensions::new(),
128        }
129    }
130}
131
132impl Request {
133    /// Build an empty request (tests / embedded). `App::handle` injects router state.
134    pub fn new(method: Method, path: impl Into<String>) -> Self {
135        Request::builder().method(method).path(path).build()
136    }
137
138    pub fn builder() -> RequestBuilder {
139        RequestBuilder {
140            method: Method::GET,
141            path: "/".into(),
142            headers: HeaderMap::new(),
143            body: Bytes::new(),
144            query: FxHashMap::default(),
145            raw_query: String::new(),
146            scheme: "http".into(),
147            host: "localhost".into(),
148            body_limit: 2 * 1024 * 1024,
149        }
150    }
151
152    /// Configured max body size (from the server / builder).
153    pub fn body_limit(&self) -> usize {
154        self.body_limit
155    }
156
157    /// Collect the full body as bytes (respecting [`Self::body_limit`]).
158    pub async fn body(&mut self) -> Result<Bytes> {
159        self.collect_body("body").await
160    }
161
162    /// Buffer the body if needed, then return UTF-8 text.
163    pub async fn text(&mut self) -> Result<String> {
164        let bytes = self.collect_body("text").await?;
165        String::from_utf8(bytes.to_vec())
166            .map_err(|e| Error::BadRequest(format!("invalid UTF-8 body: {e}")))
167    }
168
169    pub async fn json<T: DeserializeOwned>(&mut self) -> Result<T> {
170        let bytes = self.collect_body("json").await?;
171        serde_json::from_slice(&bytes).map_err(Error::from)
172    }
173
174    /// Deserialize the query string into `T`.
175    pub fn query_as<T: DeserializeOwned>(&self) -> Result<T> {
176        serde_urlencoded::from_str(&self.raw_query)
177            .map_err(|e| Error::BadRequest(format!("query error: {e}")))
178    }
179
180    /// Raw query string without leading `?` (for nested parsers like `serde_qs`).
181    pub fn raw_query(&self) -> &str {
182        &self.raw_query
183    }
184
185    pub fn query(&self, key: &str) -> Option<&str> {
186        self.query.get(key).map(|s| s.as_str())
187    }
188
189    pub fn param(&self, key: &str) -> Option<&str> {
190        self.params.get(key).map(|s| s.as_str())
191    }
192
193    /// Parse a path param (`FromStr`), or `BadRequest`.
194    pub fn param_as<T: FromStr>(&self, key: &str) -> Result<T>
195    where
196        T::Err: std::fmt::Display,
197    {
198        let raw = self
199            .param(key)
200            .ok_or_else(|| Error::BadRequest(format!("missing param `{key}`")))?;
201        raw.parse()
202            .map_err(|e| Error::BadRequest(format!("param `{key}`: {e}")))
203    }
204
205    pub fn header(&self, name: &str) -> Option<&str> {
206        self.headers.get(name).and_then(|v| v.to_str().ok())
207    }
208
209    pub fn content_type(&self) -> Option<&str> {
210        self.header("content-type")
211            .map(|v| v.split(';').next().unwrap_or(v).trim())
212    }
213
214    pub fn scheme(&self) -> &str {
215        &self.scheme
216    }
217
218    pub fn host(&self) -> &str {
219        &self.host
220    }
221
222    pub fn is_secure(&self) -> bool {
223        self.scheme.eq_ignore_ascii_case("https")
224    }
225
226    /// Absolute URL for this request path (no query).
227    pub fn url(&self) -> String {
228        if self.raw_query.is_empty() {
229            format!("{}://{}{}", self.scheme, self.host, self.path)
230        } else {
231            format!(
232                "{}://{}{}?{}",
233                self.scheme, self.host, self.path, self.raw_query
234            )
235        }
236    }
237
238    /// Take the body as a stream (once). Subsequent body reads fail.
239    pub fn into_body_stream(&mut self) -> Result<HttpBody> {
240        self.into_body_stream_as("into_body_stream")
241    }
242
243    /// Like [`Self::into_body_stream`], recording `by` in later "already consumed" errors.
244    pub fn into_body_stream_as(&mut self, by: &'static str) -> Result<HttpBody> {
245        match std::mem::replace(&mut self.body, ReqBody::Taken { by }) {
246            ReqBody::Stream(s) => Ok(s),
247            ReqBody::Bytes(b) => Ok(http_body_util::Full::new(b)
248                .map_err(|_: std::convert::Infallible| unreachable!())
249                .boxed()),
250            ReqBody::Taken { by: prev } => Err(Error::BadRequest(format!(
251                "body already consumed by {prev}"
252            ))),
253        }
254    }
255
256    pub(crate) async fn collect_body(&mut self, by: &'static str) -> Result<Bytes> {
257        match std::mem::replace(&mut self.body, ReqBody::Taken { by }) {
258            ReqBody::Bytes(b) => {
259                if b.len() > self.body_limit {
260                    return Err(Error::PayloadTooLarge);
261                }
262                self.body = ReqBody::Bytes(b.clone());
263                Ok(b)
264            }
265            ReqBody::Stream(stream) => {
266                if let Some(cl) = self
267                    .headers
268                    .get(http::header::CONTENT_LENGTH)
269                    .and_then(|v| v.to_str().ok())
270                    .and_then(|s| s.parse::<usize>().ok())
271                {
272                    if cl > self.body_limit {
273                        return Err(Error::PayloadTooLarge);
274                    }
275                }
276                let collected = collect_limited(stream, self.body_limit).await?;
277                self.body = ReqBody::Bytes(collected.clone());
278                Ok(collected)
279            }
280            ReqBody::Taken { by: prev } => Err(Error::BadRequest(format!(
281                "body already consumed by {prev}"
282            ))),
283        }
284    }
285
286    /// Shared app state. Panics if the type was never registered via `app.state`.
287    pub fn state<T>(&self) -> Arc<T>
288    where
289        T: Send + Sync + 'static,
290    {
291        self.try_state().unwrap_or_else(|| {
292            panic!(
293                "state `{}` is not registered — call app.state(..)",
294                std::any::type_name::<T>()
295            )
296        })
297    }
298
299    /// Like [`Self::state`], but returns `T::default()` when unset.
300    pub fn state_or_default<T>(&self) -> Arc<T>
301    where
302        T: Default + Send + Sync + 'static,
303    {
304        self.try_state()
305            .unwrap_or_else(|| Arc::new(T::default()))
306    }
307
308    pub fn try_state<T>(&self) -> Option<Arc<T>>
309    where
310        T: Send + Sync + 'static,
311    {
312        self.state.get::<T>()
313    }
314
315    /// Shared application [`StateMap`] (same bag as `app.state(...)`).
316    pub fn states(&self) -> Arc<StateMap> {
317        Arc::clone(&self.state)
318    }
319
320    /// Store a per-request value (e.g. from auth middleware).
321    pub fn set<T: Send + Sync + 'static>(&mut self, value: T) {
322        self.extensions.insert(value);
323    }
324
325    pub fn get<T: Send + Sync + 'static>(&self) -> Option<&T> {
326        self.extensions.get::<T>()
327    }
328
329    pub fn get_mut<T: Send + Sync + 'static>(&mut self) -> Option<&mut T> {
330        self.extensions.get_mut::<T>()
331    }
332
333    pub fn take<T: Send + Sync + 'static>(&mut self) -> Option<T> {
334        self.extensions.remove::<T>()
335    }
336
337    /// Take a pending HTTP/1 upgrade (WebSocket, …).
338    ///
339    /// Returns **503** + `Retry-After` when [`crate::App::max_upgraded_connections`]
340    /// is exhausted. Missing upgrade → `None` (not an error).
341    pub fn on_upgrade(&mut self) -> Option<std::result::Result<crate::OnUpgrade, Response>> {
342        let pending = self.take::<crate::upgrade::PendingUpgrade>()?;
343        Some(crate::upgrade::take_upgrade(pending).map_err(|b| *b))
344    }
345
346    /// Typed metadata from the matched route ([`crate::Router::with`] / plugin helpers).
347    pub fn route_meta<T>(&self) -> Option<Arc<T>>
348    where
349        T: crate::route_value::RouteValue,
350    {
351        self.get::<crate::state::MatchedMeta>()
352            .and_then(|m| m.0.get::<T>())
353    }
354
355    /// Remaining request budget from [`crate::limits::Deadline`], if set.
356    pub fn deadline_remaining(&self) -> Option<std::time::Duration> {
357        self.get::<crate::limits::Deadline>()
358            .map(|d| d.remaining())
359    }
360}
361
362/// Parse query string with `+` → space (via serde_urlencoded).
363pub fn parse_query(query: &str) -> FxHashMap<String, String> {
364    serde_urlencoded::from_str::<FxHashMap<String, String>>(query).unwrap_or_default()
365}
366
367pub fn percent_decode(input: &str) -> String {
368    if !input.as_bytes().contains(&b'%') {
369        return input.to_string();
370    }
371    percent_encoding::percent_decode_str(input)
372        .decode_utf8_lossy()
373        .into_owned()
374}
375
376/// Build scheme/host from the incoming request (and proxy headers when trusted).
377pub(crate) fn resolve_scheme_host(
378    headers: &HeaderMap,
379    uri_scheme: Option<&str>,
380    trust_proxy: bool,
381) -> (String, String) {
382    let mut scheme = uri_scheme.unwrap_or("http").to_string();
383    let mut host = headers
384        .get(http::header::HOST)
385        .and_then(|v| v.to_str().ok())
386        .unwrap_or("localhost")
387        .to_string();
388
389    if trust_proxy {
390        if let Some(proto) = headers
391            .get("x-forwarded-proto")
392            .and_then(|v| v.to_str().ok())
393            .and_then(|s| s.split(',').next())
394            .map(str::trim)
395            .filter(|s| !s.is_empty())
396        {
397            scheme = proto.to_string();
398        }
399        if let Some(h) = headers
400            .get("x-forwarded-host")
401            .and_then(|v| v.to_str().ok())
402            .and_then(|s| s.split(',').next())
403            .map(str::trim)
404            .filter(|s| !s.is_empty())
405        {
406            host = h.to_string();
407        }
408    }
409    (scheme, host)
410}
sova_core/request/mod.rs

sova_core/request/
mod.rs
