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