millipede_core/
request.rs

1//! Request values and their builder API.
2
3use serde::{Deserialize, Serialize};
4use std::fmt;
5use url::Url;
6
7pub use http::{HeaderMap, Method};
8
9/// A crawl request and its processing state.
10#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
11pub struct Request {
12    /// Stable identifier derived from `unique_key`.
13    pub id: RequestId,
14    /// URL to navigate to.
15    pub url: Url,
16    /// Final URL after redirects, when navigation has occurred.
17    pub loaded_url: Option<Url>,
18    /// Queue deduplication key.
19    pub unique_key: String,
20    /// HTTP method.
21    #[serde(with = "method_serde")]
22    pub method: Method,
23    /// HTTP headers, including repeated values.
24    #[serde(with = "header_map_serde")]
25    pub headers: HeaderMap,
26    /// Optional request body.
27    pub body: Option<RequestBody>,
28    /// User-defined structured metadata.
29    pub user_data: UserData,
30    /// Optional routing label.
31    pub label: Option<String>,
32    /// Number of retry attempts made.
33    pub retry_count: u32,
34    /// Number of session rotations made.
35    pub session_rotation_count: u32,
36    /// Per-request retry limit override.
37    pub max_retries: Option<u32>,
38    /// Whether retries are disabled.
39    pub no_retry: bool,
40    /// Errors recorded during processing.
41    pub error_messages: Vec<String>,
42    /// Time at which processing completed.
43    #[serde(with = "time::serde::rfc3339::option")]
44    pub handled_at: Option<time::OffsetDateTime>,
45    /// Current lifecycle state.
46    pub state: RequestState,
47    /// Depth at which this request was discovered.
48    pub crawl_depth: u32,
49    /// Whether navigation should be skipped.
50    pub skip_navigation: bool,
51    // `enqueue_strategy` is intentionally omitted until ROADMAP Phase 5.
52}
53
54/// The processing lifecycle state of a request.
55#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
56pub enum RequestState {
57    /// Waiting to be processed.
58    #[default]
59    Unprocessed,
60    /// Running pre-navigation hooks.
61    BeforeNav,
62    /// Running post-navigation hooks.
63    AfterNav,
64    /// Running the request handler.
65    RequestHandler,
66    /// Successfully completed.
67    Done,
68    /// Running the error handler.
69    ErrorHandler,
70    /// Permanently failed.
71    Error,
72    /// Deliberately skipped.
73    Skipped,
74}
75
76/// A deterministic request identifier.
77#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
78pub struct RequestId(String);
79
80impl RequestId {
81    /// Derives an identifier deterministically from a request unique key.
82    pub fn from_unique_key(unique_key: &str) -> Self {
83        let h1 = fnv1a64(unique_key.as_bytes());
84        let h2 = fnv1a64_seeded(unique_key.as_bytes(), h1);
85        Self(format!("{h1:016x}{h2:016x}"))
86    }
87
88    /// Returns the identifier as a string slice.
89    pub fn as_str(&self) -> &str {
90        &self.0
91    }
92}
93
94impl fmt::Display for RequestId {
95    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
96        formatter.write_str(&self.0)
97    }
98}
99
100/// A supported request-body representation.
101#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
102#[serde(rename_all = "snake_case")]
103pub enum RequestBody {
104    /// Uninterpreted bytes.
105    Bytes(Vec<u8>),
106    /// Ordered form key-value pairs.
107    Form(Vec<(String, String)>),
108    /// A JSON value.
109    Json(serde_json::Value),
110}
111
112impl RequestBody {
113    /// Returns the canonical bytes used for unique-key hashing.
114    pub fn canonical_bytes(&self) -> Vec<u8> {
115        match self {
116            Self::Bytes(bytes) => bytes.clone(),
117            Self::Form(pairs) => pairs
118                .iter()
119                .map(|(key, value)| format!("{key}={value}"))
120                .collect::<Vec<_>>()
121                .join("&")
122                .into_bytes(),
123            Self::Json(value) => serde_json::to_vec(value).expect(
124                // Serialization can only fail for non-string map keys, which Value forbids.
125                "a serde_json::Value is always serializable",
126            ),
127        }
128    }
129}
130
131/// User-defined JSON metadata attached to a request.
132#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
133pub struct UserData(pub serde_json::Map<String, serde_json::Value>);
134
135impl UserData {
136    /// Deserializes the value at `key`, returning `None` when it is absent.
137    pub fn get_typed<T: serde::de::DeserializeOwned>(
138        &self,
139        key: &str,
140    ) -> Option<Result<T, serde_json::Error>> {
141        self.0.get(key).cloned().map(serde_json::from_value)
142    }
143
144    /// Serializes and stores a typed value at `key`.
145    pub fn set_typed<T: Serialize>(
146        &mut self,
147        key: &str,
148        value: &T,
149    ) -> Result<(), serde_json::Error> {
150        self.0.insert(key.to_owned(), serde_json::to_value(value)?);
151        Ok(())
152    }
153
154    /// Returns whether no user data is stored.
155    pub fn is_empty(&self) -> bool {
156        self.0.is_empty()
157    }
158
159    /// Returns the raw JSON value stored at `key`.
160    pub fn get(&self, key: &str) -> Option<&serde_json::Value> {
161        self.0.get(key)
162    }
163}
164
165impl Request {
166    /// Creates a builder configured for a GET request.
167    pub fn get(url: impl IntoUrl) -> RequestBuilder {
168        Self::builder().url(url).method(Method::GET)
169    }
170
171    /// Creates a builder configured for a POST request.
172    pub fn post(url: impl IntoUrl) -> RequestBuilder {
173        Self::builder().url(url).method(Method::POST)
174    }
175
176    /// Creates an empty request builder.
177    pub fn builder() -> RequestBuilder {
178        RequestBuilder::default()
179    }
180
181    /// Computes a deterministic queue deduplication key.
182    pub fn compute_unique_key(url: &Url, method: &Method, body: Option<&RequestBody>) -> String {
183        let mut normalized = url.clone();
184        normalized.set_fragment(None);
185        if *method == Method::GET && body.is_none() {
186            normalized.into()
187        } else {
188            let body_bytes = body.map(RequestBody::canonical_bytes).unwrap_or_default();
189            format!(
190                "{}({:016x}):{}",
191                method.as_str(),
192                fnv1a64(&body_bytes),
193                normalized
194            )
195        }
196    }
197}
198
199/// Conversion into a parsed URL for request builders.
200pub trait IntoUrl {
201    /// Converts this value into a URL.
202    fn into_url(self) -> Result<Url, url::ParseError>;
203}
204
205impl IntoUrl for Url {
206    fn into_url(self) -> Result<Url, url::ParseError> {
207        Ok(self)
208    }
209}
210
211impl IntoUrl for &str {
212    fn into_url(self) -> Result<Url, url::ParseError> {
213        Url::parse(self)
214    }
215}
216
217impl IntoUrl for String {
218    fn into_url(self) -> Result<Url, url::ParseError> {
219        Url::parse(&self)
220    }
221}
222
223impl IntoUrl for &String {
224    fn into_url(self) -> Result<Url, url::ParseError> {
225        Url::parse(self)
226    }
227}
228
229#[derive(Debug)]
230struct PendingHeader {
231    name: String,
232    value: String,
233}
234
235/// Builds a [`Request`] while deferring parsing and serialization errors.
236#[derive(Debug, Default)]
237#[must_use = "builders do nothing unless consumed by build"]
238pub struct RequestBuilder {
239    url: Option<Result<Url, url::ParseError>>,
240    method: Option<Method>,
241    headers: HeaderMap,
242    pending_headers: Vec<PendingHeader>,
243    body: Option<RequestBody>,
244    serialization_error: Option<serde_json::Error>,
245    user_data: UserData,
246    label: Option<String>,
247    max_retries: Option<u32>,
248    no_retry: bool,
249    skip_navigation: bool,
250    unique_key: Option<String>,
251    pub(crate) forefront: bool,
252    crawl_depth: u32,
253}
254
255impl RequestBuilder {
256    /// Builds and adds this request to a queue.
257    pub async fn enqueue(
258        self,
259        queue: &dyn crate::storage::RequestQueue,
260    ) -> Result<crate::storage::QueueOpInfo, crate::errors::CrawlError> {
261        let forefront = self.is_forefront();
262        let request = self.build()?;
263        Ok(queue
264            .add(
265                request,
266                crate::storage::AddOptions {
267                    forefront,
268                    ..Default::default()
269                },
270            )
271            .await?)
272    }
273
274    /// Sets the request URL.
275    pub fn url(mut self, url: impl IntoUrl) -> Self {
276        self.url = Some(url.into_url());
277        self
278    }
279
280    /// Sets the HTTP method.
281    pub fn method(mut self, method: Method) -> Self {
282        self.method = Some(method);
283        self
284    }
285
286    /// Appends a header, deferring validation until [`Self::build`].
287    pub fn header(mut self, name: &str, value: &str) -> Self {
288        self.pending_headers.push(PendingHeader {
289            name: name.to_owned(),
290            value: value.to_owned(),
291        });
292        self
293    }
294
295    /// Replaces all currently configured headers.
296    pub fn headers(mut self, headers: HeaderMap) -> Self {
297        self.headers = headers;
298        self.pending_headers.clear();
299        self
300    }
301
302    /// Sets the request body.
303    pub fn body(mut self, body: RequestBody) -> Self {
304        self.body = Some(body);
305        self
306    }
307
308    /// Serializes and sets a JSON request body.
309    pub fn json<T: Serialize>(mut self, value: &T) -> Self {
310        match serde_json::to_value(value) {
311            Ok(value) => self.body = Some(RequestBody::Json(value)),
312            Err(error) => self.serialization_error = Some(error),
313        }
314        self
315    }
316
317    /// Sets an ordered form request body.
318    pub fn form(mut self, pairs: impl IntoIterator<Item = (String, String)>) -> Self {
319        self.body = Some(RequestBody::Form(pairs.into_iter().collect()));
320        self
321    }
322
323    /// Sets the routing label.
324    pub fn label(mut self, label: impl Into<String>) -> Self {
325        self.label = Some(label.into());
326        self
327    }
328
329    /// Replaces the request's user data.
330    pub fn user_data(mut self, user_data: UserData) -> Self {
331        self.user_data = user_data;
332        self
333    }
334
335    /// Serializes and inserts an entry into user data.
336    pub fn user_data_entry<T: Serialize>(mut self, key: impl Into<String>, value: &T) -> Self {
337        match serde_json::to_value(value) {
338            Ok(value) => {
339                self.user_data.0.insert(key.into(), value);
340            }
341            Err(error) => self.serialization_error = Some(error),
342        }
343        self
344    }
345
346    /// Sets the per-request retry limit.
347    pub fn max_retries(mut self, max_retries: u32) -> Self {
348        self.max_retries = Some(max_retries);
349        self
350    }
351
352    /// Enables or disables retries.
353    pub fn no_retry(mut self, no_retry: bool) -> Self {
354        self.no_retry = no_retry;
355        self
356    }
357
358    /// Controls whether navigation is skipped.
359    pub fn skip_navigation(mut self, skip_navigation: bool) -> Self {
360        self.skip_navigation = skip_navigation;
361        self
362    }
363
364    /// Overrides the computed queue deduplication key.
365    pub fn unique_key(mut self, unique_key: impl Into<String>) -> Self {
366        self.unique_key = Some(unique_key.into());
367        self
368    }
369
370    /// Controls whether a future queue enqueue places this request at the front.
371    pub fn forefront(mut self, forefront: bool) -> Self {
372        self.forefront = forefront;
373        self
374    }
375
376    /// Returns whether future queue enqueue convenience should use the forefront.
377    pub fn is_forefront(&self) -> bool {
378        self.forefront
379    }
380
381    /// Sets the discovery depth.
382    pub fn crawl_depth(mut self, crawl_depth: u32) -> Self {
383        self.crawl_depth = crawl_depth;
384        self
385    }
386
387    /// Validates this builder and creates a request.
388    pub fn build(mut self) -> Result<Request, RequestBuildError> {
389        if let Some(error) = self.serialization_error {
390            return Err(error.into());
391        }
392        let url = self.url.ok_or(RequestBuildError::MissingUrl)??;
393        for pending in self.pending_headers {
394            let name =
395                http::header::HeaderName::from_bytes(pending.name.as_bytes()).map_err(|error| {
396                    RequestBuildError::InvalidHeader {
397                        name: pending.name.clone(),
398                        message: error.to_string(),
399                    }
400                })?;
401            let value =
402                http::HeaderValue::from_bytes(pending.value.as_bytes()).map_err(|error| {
403                    RequestBuildError::InvalidHeader {
404                        name: pending.name.clone(),
405                        message: error.to_string(),
406                    }
407                })?;
408            self.headers.append(name, value);
409        }
410        let method = self.method.unwrap_or(Method::GET);
411        let unique_key = self
412            .unique_key
413            .unwrap_or_else(|| Request::compute_unique_key(&url, &method, self.body.as_ref()));
414        let id = RequestId::from_unique_key(&unique_key);
415        Ok(Request {
416            id,
417            url,
418            loaded_url: None,
419            unique_key,
420            method,
421            headers: self.headers,
422            body: self.body,
423            user_data: self.user_data,
424            label: self.label,
425            retry_count: 0,
426            session_rotation_count: 0,
427            max_retries: self.max_retries,
428            no_retry: self.no_retry,
429            error_messages: Vec::new(),
430            handled_at: None,
431            state: RequestState::Unprocessed,
432            crawl_depth: self.crawl_depth,
433            skip_navigation: self.skip_navigation,
434        })
435    }
436}
437
438/// Errors encountered while building a request.
439#[derive(Debug, thiserror::Error)]
440#[non_exhaustive]
441pub enum RequestBuildError {
442    /// The builder has no URL.
443    #[error("request URL is missing")]
444    MissingUrl,
445    /// The supplied URL could not be parsed.
446    #[error("invalid URL: {0}")]
447    InvalidUrl(#[from] url::ParseError),
448    /// A header name or value is invalid.
449    #[error("invalid header {name}: {message}")]
450    InvalidHeader {
451        /// Invalid header name as supplied by the caller.
452        name: String,
453        /// Parser error describing the invalid header.
454        message: String,
455    },
456    /// JSON serialization failed.
457    #[error("user data serialization failed: {0}")]
458    Serialization(#[from] serde_json::Error),
459}
460
461fn fnv1a64(bytes: &[u8]) -> u64 {
462    fnv1a64_seeded(bytes, 0xcbf29ce484222325)
463}
464
465fn fnv1a64_seeded(bytes: &[u8], seed: u64) -> u64 {
466    bytes.iter().fold(seed, |hash, byte| {
467        (hash ^ u64::from(*byte)).wrapping_mul(0x100000001b3)
468    })
469}
470
471mod method_serde {
472    use http::Method;
473    use serde::{Deserialize, Deserializer, Serializer, de::Error as _};
474
475    pub fn serialize<S: Serializer>(method: &Method, serializer: S) -> Result<S::Ok, S::Error> {
476        serializer.serialize_str(method.as_str())
477    }
478
479    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Method, D::Error> {
480        let value = String::deserialize(deserializer)?;
481        Method::from_bytes(value.as_bytes()).map_err(D::Error::custom)
482    }
483}
484
485mod header_map_serde {
486    use http::{HeaderMap, HeaderName, HeaderValue};
487    use serde::{
488        Deserialize, Deserializer, Serialize, Serializer, de::Error as _, ser::Error as _,
489    };
490
491    pub fn serialize<S: Serializer>(headers: &HeaderMap, serializer: S) -> Result<S::Ok, S::Error> {
492        headers
493            .iter()
494            .map(|(name, value)| {
495                Ok((
496                    name.as_str().to_owned(),
497                    value.to_str().map_err(S::Error::custom)?.to_owned(),
498                ))
499            })
500            .collect::<Result<Vec<(String, String)>, S::Error>>()?
501            .serialize(serializer)
502    }
503
504    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<HeaderMap, D::Error> {
505        let pairs = Vec::<(String, String)>::deserialize(deserializer)?;
506        let mut headers = HeaderMap::new();
507        for (name, value) in pairs {
508            let name = HeaderName::from_bytes(name.as_bytes()).map_err(D::Error::custom)?;
509            let value = HeaderValue::from_bytes(value.as_bytes()).map_err(D::Error::custom)?;
510            headers.append(name, value);
511        }
512        Ok(headers)
513    }
514}
515
516#[cfg(test)]
517mod tests {
518    use super::*;
519
520    #[test]
521    fn builder_happy_path() {
522        let request = Request::get("https://example.com/a?x=1")
523            .label("l")
524            .build()
525            .unwrap();
526        assert_eq!(request.label.as_deref(), Some("l"));
527        assert_eq!(request.method, Method::GET);
528    }
529
530    #[test]
531    fn unique_key_override_is_respected() {
532        let request = Request::get("https://example.com")
533            .unique_key("mine")
534            .build()
535            .unwrap();
536        assert_eq!(request.unique_key, "mine");
537    }
538
539    #[test]
540    fn get_key_is_fragmentless_url() {
541        let request = Request::get("https://example.com/a#frag").build().unwrap();
542        assert_eq!(request.unique_key, "https://example.com/a");
543    }
544
545    #[test]
546    fn fragment_does_not_affect_key() {
547        let first = Request::get("https://example.com/a#frag").build().unwrap();
548        let second = Request::get("https://example.com/a").build().unwrap();
549        assert_eq!(first.unique_key, second.unique_key);
550    }
551
552    #[test]
553    fn post_body_affects_key_deterministically() {
554        let build = |bytes| {
555            Request::post("https://example.com/a")
556                .body(RequestBody::Bytes(bytes))
557                .build()
558                .unwrap()
559        };
560        assert_eq!(build(vec![1]).unique_key, build(vec![1]).unique_key);
561        assert_ne!(build(vec![1]).unique_key, build(vec![2]).unique_key);
562    }
563
564    #[test]
565    fn request_id_is_deterministic() {
566        assert_eq!(
567            RequestId::from_unique_key("key"),
568            RequestId::from_unique_key("key")
569        );
570    }
571
572    #[test]
573    fn user_data_typed_roundtrip() {
574        #[derive(Debug, PartialEq, Serialize, Deserialize)]
575        struct Example {
576            number: u32,
577        }
578
579        let mut data = UserData::default();
580        data.set_typed("example", &Example { number: 7 }).unwrap();
581        assert_eq!(
582            data.get_typed::<Example>("example").unwrap().unwrap(),
583            Example { number: 7 }
584        );
585    }
586
587    #[test]
588    fn invalid_header_is_reported_at_build() {
589        let result = Request::get("https://example.com")
590            .header("bad header", "value")
591            .build();
592        assert!(matches!(
593            result,
594            Err(RequestBuildError::InvalidHeader { .. })
595        ));
596    }
597}
millipede_core/request.rs

millipede_core/
request.rs
