qstash_rs/client/
request.rs

1use std::fmt;
2
3use bytes::Bytes;
4use reqwest::{
5    header::{HeaderMap, HeaderName, HeaderValue, CONTENT_TYPE},
6    Method, Url,
7};
8use serde::Serialize;
9
10use crate::error::{Error, Result};
11
12/// A QStash destination.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub enum Destination {
15    /// Deliver to a concrete URL.
16    Url(Url),
17    /// Deliver to a URL Group.
18    UrlGroup(String),
19}
20
21impl Destination {
22    /// Parses a concrete HTTP destination URL.
23    pub fn url(url: impl AsRef<str>) -> Result<Self> {
24        Url::parse(url.as_ref())
25            .map(Self::Url)
26            .map_err(|error| Error::Config {
27                message: format!("invalid destination url: {error}"),
28            })
29    }
30
31    /// Creates a URL Group destination.
32    pub fn url_group(name: impl Into<String>) -> Self {
33        Self::UrlGroup(name.into())
34    }
35
36    pub(crate) fn path_value(&self) -> String {
37        match self {
38            Self::Url(url) => url.as_str().to_owned(),
39            Self::UrlGroup(name) => name.clone(),
40        }
41    }
42}
43
44impl fmt::Display for Destination {
45    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
46        match self {
47            Self::Url(url) => formatter.write_str(url.as_str()),
48            Self::UrlGroup(name) => formatter.write_str(name),
49        }
50    }
51}
52
53/// Redaction settings sent to QStash.
54#[derive(Debug, Clone, Default, PartialEq, Eq)]
55pub struct Redaction {
56    body: bool,
57    all_headers: bool,
58    header_names: Vec<String>,
59}
60
61impl Redaction {
62    /// Creates an empty redaction configuration.
63    pub fn new() -> Self {
64        Self::default()
65    }
66
67    /// Redacts the message body from logs and the dashboard.
68    pub fn body(mut self) -> Self {
69        self.body = true;
70        self
71    }
72
73    /// Redacts all request headers from logs and the dashboard.
74    pub fn all_headers(mut self) -> Self {
75        self.all_headers = true;
76        self
77    }
78
79    /// Redacts a single header from logs and the dashboard.
80    pub fn header(mut self, name: impl Into<String>) -> Self {
81        self.header_names.push(name.into());
82        self
83    }
84
85    pub(crate) fn header_value(&self) -> Option<String> {
86        let mut parts = Vec::new();
87
88        if self.body {
89            parts.push(String::from("body"));
90        }
91
92        if self.all_headers {
93            parts.push(String::from("header"));
94        }
95
96        parts.extend(
97            self.header_names
98                .iter()
99                .map(|name| format!("header[{name}]")),
100        );
101
102        if parts.is_empty() {
103            None
104        } else {
105            Some(parts.join(","))
106        }
107    }
108}
109
110/// A normalized publish response item.
111#[derive(Debug, Clone, serde::Deserialize, PartialEq, Eq)]
112#[serde(rename_all = "camelCase")]
113pub struct PublishedMessage {
114    /// The created or deduplicated message identifier.
115    pub message_id: Option<String>,
116    /// Destination URL when the response includes one.
117    pub url: Option<String>,
118    /// Error returned by QStash for this item.
119    pub error: Option<String>,
120    /// Whether the publish request was deduplicated.
121    pub deduplicated: Option<bool>,
122}
123
124/// Publish response normalized to a list for both single-URL and URL Group publishes.
125pub type PublishResponse = Vec<PublishedMessage>;
126
127/// A publish request.
128#[derive(Debug, Clone)]
129pub struct PublishRequest {
130    pub(crate) destination: Destination,
131    pub(crate) body: Option<Bytes>,
132    pub(crate) headers: HeaderMap,
133    pub(crate) method: Method,
134    pub(crate) delay: Option<u32>,
135    pub(crate) not_before: Option<u64>,
136    pub(crate) deduplication_id: Option<String>,
137    pub(crate) content_based_deduplication: bool,
138    pub(crate) retries: Option<u32>,
139    pub(crate) retry_delay: Option<String>,
140    pub(crate) callback: Option<String>,
141    pub(crate) failure_callback: Option<String>,
142    pub(crate) timeout: Option<u32>,
143    pub(crate) label: Option<String>,
144    pub(crate) redaction: Option<Redaction>,
145}
146
147impl PublishRequest {
148    /// Starts a builder for a publish request.
149    pub fn builder(destination: Destination) -> PublishRequestBuilder {
150        PublishRequestBuilder::new(destination)
151    }
152
153    /// Returns the destination.
154    pub fn destination(&self) -> &Destination {
155        &self.destination
156    }
157}
158
159/// Builder for [`PublishRequest`].
160#[derive(Debug, Clone)]
161pub struct PublishRequestBuilder {
162    request: PublishRequest,
163}
164
165impl PublishRequestBuilder {
166    fn new(destination: Destination) -> Self {
167        Self {
168            request: PublishRequest {
169                destination,
170                body: None,
171                headers: HeaderMap::new(),
172                method: Method::POST,
173                delay: None,
174                not_before: None,
175                deduplication_id: None,
176                content_based_deduplication: false,
177                retries: None,
178                retry_delay: None,
179                callback: None,
180                failure_callback: None,
181                timeout: None,
182                label: None,
183                redaction: None,
184            },
185        }
186    }
187
188    /// Sets a raw request body.
189    pub fn body(mut self, body: impl Into<Bytes>) -> Self {
190        self.request.body = Some(body.into());
191        self
192    }
193
194    /// Serializes a JSON body and sets the content type.
195    pub fn json_body<T>(mut self, body: &T) -> Result<Self>
196    where
197        T: Serialize,
198    {
199        self.request.body = Some(Bytes::from(
200            serde_json::to_vec(body).map_err(Error::Serialize)?,
201        ));
202        self.request
203            .headers
204            .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
205        Ok(self)
206    }
207
208    /// Adds a single header that should be forwarded to the destination.
209    pub fn header(mut self, name: &str, value: &str) -> Result<Self> {
210        let header_name =
211            HeaderName::from_bytes(name.as_bytes()).map_err(|error| Error::InvalidRequest {
212                message: format!("invalid header name `{name}`: {error}"),
213            })?;
214        let header_value = HeaderValue::from_str(value).map_err(|error| Error::InvalidRequest {
215            message: format!("invalid header value for `{name}`: {error}"),
216        })?;
217        self.request.headers.insert(header_name, header_value);
218        Ok(self)
219    }
220
221    /// Replaces the destination headers.
222    pub fn headers(mut self, headers: HeaderMap) -> Self {
223        self.request.headers = headers;
224        self
225    }
226
227    /// Overrides the delivery method.
228    pub fn method(mut self, method: Method) -> Self {
229        self.request.method = method;
230        self
231    }
232
233    /// Delays delivery by the provided number of seconds.
234    pub fn delay_seconds(mut self, seconds: u32) -> Self {
235        self.request.delay = Some(seconds);
236        self
237    }
238
239    /// Sets an absolute delivery time expressed as a Unix timestamp in seconds.
240    pub fn not_before(mut self, timestamp: u64) -> Self {
241        self.request.not_before = Some(timestamp);
242        self
243    }
244
245    /// Sets a manual deduplication identifier.
246    pub fn deduplication_id(mut self, id: impl Into<String>) -> Self {
247        self.request.deduplication_id = Some(id.into());
248        self
249    }
250
251    /// Enables content-based deduplication.
252    pub fn content_based_deduplication(mut self, enabled: bool) -> Self {
253        self.request.content_based_deduplication = enabled;
254        self
255    }
256
257    /// Sets the retry count.
258    pub fn retries(mut self, retries: u32) -> Self {
259        self.request.retries = Some(retries);
260        self
261    }
262
263    /// Sets the retry delay expression.
264    pub fn retry_delay(mut self, expression: impl Into<String>) -> Self {
265        self.request.retry_delay = Some(expression.into());
266        self
267    }
268
269    /// Sets the callback URL.
270    pub fn callback(mut self, url: impl Into<String>) -> Self {
271        self.request.callback = Some(url.into());
272        self
273    }
274
275    /// Sets the failure callback URL.
276    pub fn failure_callback(mut self, url: impl Into<String>) -> Self {
277        self.request.failure_callback = Some(url.into());
278        self
279    }
280
281    /// Sets a shorter destination timeout in seconds.
282    pub fn timeout_seconds(mut self, seconds: u32) -> Self {
283        self.request.timeout = Some(seconds);
284        self
285    }
286
287    /// Assigns a label that can later be used to filter logs and DLQ entries.
288    pub fn label(mut self, label: impl Into<String>) -> Self {
289        self.request.label = Some(label.into());
290        self
291    }
292
293    /// Configures redaction for the publish request.
294    pub fn redaction(mut self, redaction: Redaction) -> Self {
295        self.request.redaction = Some(redaction);
296        self
297    }
298
299    /// Finishes the builder.
300    pub fn build(self) -> PublishRequest {
301        self.request
302    }
303}
304
305/// A batch publish item.
306#[derive(Debug, Clone)]
307pub struct BatchRequest {
308    pub(crate) publish: PublishRequest,
309    pub(crate) queue_name: Option<String>,
310}
311
312impl BatchRequest {
313    /// Creates a batch item from a publish request.
314    pub fn new(publish: PublishRequest) -> Self {
315        Self {
316            publish,
317            queue_name: None,
318        }
319    }
320
321    /// Enqueues the batch item into a queue instead of publishing it directly.
322    pub fn queue_name(mut self, queue_name: impl Into<String>) -> Self {
323        self.queue_name = Some(queue_name.into());
324        self
325    }
326}
327
328/// A schedule creation request.
329#[derive(Debug, Clone)]
330pub struct ScheduleRequest {
331    pub(crate) destination: Destination,
332    pub(crate) cron: String,
333    pub(crate) schedule_id: Option<String>,
334    pub(crate) body: Option<Bytes>,
335    pub(crate) headers: HeaderMap,
336    pub(crate) method: Method,
337    pub(crate) delay: Option<u32>,
338    pub(crate) retries: Option<u32>,
339    pub(crate) retry_delay: Option<String>,
340    pub(crate) callback: Option<String>,
341    pub(crate) failure_callback: Option<String>,
342    pub(crate) timeout: Option<u32>,
343    pub(crate) queue_name: Option<String>,
344    pub(crate) label: Option<String>,
345    pub(crate) redaction: Option<Redaction>,
346}
347
348impl ScheduleRequest {
349    /// Starts a builder for a schedule request.
350    pub fn builder(destination: Destination, cron: impl Into<String>) -> ScheduleRequestBuilder {
351        ScheduleRequestBuilder::new(destination, cron.into())
352    }
353}
354
355/// Builder for [`ScheduleRequest`].
356#[derive(Debug, Clone)]
357pub struct ScheduleRequestBuilder {
358    request: ScheduleRequest,
359}
360
361impl ScheduleRequestBuilder {
362    fn new(destination: Destination, cron: String) -> Self {
363        Self {
364            request: ScheduleRequest {
365                destination,
366                cron,
367                schedule_id: None,
368                body: None,
369                headers: HeaderMap::new(),
370                method: Method::POST,
371                delay: None,
372                retries: None,
373                retry_delay: None,
374                callback: None,
375                failure_callback: None,
376                timeout: None,
377                queue_name: None,
378                label: None,
379                redaction: None,
380            },
381        }
382    }
383
384    /// Reuses the provided schedule identifier.
385    pub fn schedule_id(mut self, schedule_id: impl Into<String>) -> Self {
386        self.request.schedule_id = Some(schedule_id.into());
387        self
388    }
389
390    /// Sets a raw request body.
391    pub fn body(mut self, body: impl Into<Bytes>) -> Self {
392        self.request.body = Some(body.into());
393        self
394    }
395
396    /// Serializes a JSON body and sets the content type.
397    pub fn json_body<T>(mut self, body: &T) -> Result<Self>
398    where
399        T: Serialize,
400    {
401        self.request.body = Some(Bytes::from(
402            serde_json::to_vec(body).map_err(Error::Serialize)?,
403        ));
404        self.request
405            .headers
406            .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
407        Ok(self)
408    }
409
410    /// Adds a single header that should be forwarded to the destination.
411    pub fn header(mut self, name: &str, value: &str) -> Result<Self> {
412        let header_name =
413            HeaderName::from_bytes(name.as_bytes()).map_err(|error| Error::InvalidRequest {
414                message: format!("invalid header name `{name}`: {error}"),
415            })?;
416        let header_value = HeaderValue::from_str(value).map_err(|error| Error::InvalidRequest {
417            message: format!("invalid header value for `{name}`: {error}"),
418        })?;
419        self.request.headers.insert(header_name, header_value);
420        Ok(self)
421    }
422
423    /// Replaces the destination headers.
424    pub fn headers(mut self, headers: HeaderMap) -> Self {
425        self.request.headers = headers;
426        self
427    }
428
429    /// Overrides the delivery method.
430    pub fn method(mut self, method: Method) -> Self {
431        self.request.method = method;
432        self
433    }
434
435    /// Adds a post-trigger delay in seconds.
436    pub fn delay_seconds(mut self, seconds: u32) -> Self {
437        self.request.delay = Some(seconds);
438        self
439    }
440
441    /// Sets the retry count.
442    pub fn retries(mut self, retries: u32) -> Self {
443        self.request.retries = Some(retries);
444        self
445    }
446
447    /// Sets the retry delay expression.
448    pub fn retry_delay(mut self, expression: impl Into<String>) -> Self {
449        self.request.retry_delay = Some(expression.into());
450        self
451    }
452
453    /// Sets the callback URL.
454    pub fn callback(mut self, url: impl Into<String>) -> Self {
455        self.request.callback = Some(url.into());
456        self
457    }
458
459    /// Sets the failure callback URL.
460    pub fn failure_callback(mut self, url: impl Into<String>) -> Self {
461        self.request.failure_callback = Some(url.into());
462        self
463    }
464
465    /// Sets a shorter destination timeout in seconds.
466    pub fn timeout_seconds(mut self, seconds: u32) -> Self {
467        self.request.timeout = Some(seconds);
468        self
469    }
470
471    /// Sends each scheduled execution through a queue.
472    pub fn queue_name(mut self, queue_name: impl Into<String>) -> Self {
473        self.request.queue_name = Some(queue_name.into());
474        self
475    }
476
477    /// Assigns a label that can later be used to filter logs and DLQ entries.
478    pub fn label(mut self, label: impl Into<String>) -> Self {
479        self.request.label = Some(label.into());
480        self
481    }
482
483    /// Configures redaction for the scheduled request.
484    pub fn redaction(mut self, redaction: Redaction) -> Self {
485        self.request.redaction = Some(redaction);
486        self
487    }
488
489    /// Finishes the builder.
490    pub fn build(self) -> ScheduleRequest {
491        self.request
492    }
493}
494
495pub(crate) fn build_publish_headers(request: &PublishRequest) -> Result<HeaderMap> {
496    let mut headers = prefix_forward_headers(&request.headers)?;
497    headers.insert("Upstash-Method", header_value(request.method.as_str())?);
498
499    if let Some(delay) = request.delay {
500        headers.insert("Upstash-Delay", header_value(&format!("{delay}s"))?);
501    }
502
503    if let Some(not_before) = request.not_before {
504        headers.insert("Upstash-Not-Before", header_value(&not_before.to_string())?);
505    }
506
507    if let Some(deduplication_id) = &request.deduplication_id {
508        headers.insert("Upstash-Deduplication-Id", header_value(deduplication_id)?);
509    }
510
511    if request.content_based_deduplication {
512        headers.insert(
513            "Upstash-Content-Based-Deduplication",
514            HeaderValue::from_static("true"),
515        );
516    }
517
518    if let Some(retries) = request.retries {
519        headers.insert("Upstash-Retries", header_value(&retries.to_string())?);
520    }
521
522    if let Some(retry_delay) = &request.retry_delay {
523        headers.insert("Upstash-Retry-Delay", header_value(retry_delay)?);
524    }
525
526    if let Some(callback) = &request.callback {
527        headers.insert("Upstash-Callback", header_value(callback)?);
528    }
529
530    if let Some(failure_callback) = &request.failure_callback {
531        headers.insert("Upstash-Failure-Callback", header_value(failure_callback)?);
532    }
533
534    if let Some(timeout) = request.timeout {
535        headers.insert("Upstash-Timeout", header_value(&format!("{timeout}s"))?);
536    }
537
538    if let Some(label) = &request.label {
539        headers.insert("Upstash-Label", header_value(label)?);
540    }
541
542    if let Some(redaction) = &request.redaction {
543        if let Some(value) = redaction.header_value() {
544            headers.insert("Upstash-Redact-Fields", header_value(&value)?);
545        }
546    }
547
548    Ok(headers)
549}
550
551pub(crate) fn build_schedule_headers(request: &ScheduleRequest) -> Result<HeaderMap> {
552    let mut headers = prefix_forward_headers(&request.headers)?;
553
554    headers.insert("Upstash-Cron", header_value(&request.cron)?);
555    headers.insert("Upstash-Method", header_value(request.method.as_str())?);
556
557    if let Some(schedule_id) = &request.schedule_id {
558        headers.insert("Upstash-Schedule-Id", header_value(schedule_id)?);
559    }
560
561    if let Some(delay) = request.delay {
562        headers.insert("Upstash-Delay", header_value(&format!("{delay}s"))?);
563    }
564
565    if let Some(retries) = request.retries {
566        headers.insert("Upstash-Retries", header_value(&retries.to_string())?);
567    }
568
569    if let Some(retry_delay) = &request.retry_delay {
570        headers.insert("Upstash-Retry-Delay", header_value(retry_delay)?);
571    }
572
573    if let Some(callback) = &request.callback {
574        headers.insert("Upstash-Callback", header_value(callback)?);
575    }
576
577    if let Some(failure_callback) = &request.failure_callback {
578        headers.insert("Upstash-Failure-Callback", header_value(failure_callback)?);
579    }
580
581    if let Some(timeout) = request.timeout {
582        headers.insert("Upstash-Timeout", header_value(&format!("{timeout}s"))?);
583    }
584
585    if let Some(queue_name) = &request.queue_name {
586        headers.insert("Upstash-Queue-Name", header_value(queue_name)?);
587    }
588
589    if let Some(label) = &request.label {
590        headers.insert("Upstash-Label", header_value(label)?);
591    }
592
593    if let Some(redaction) = &request.redaction {
594        if let Some(value) = redaction.header_value() {
595            headers.insert("Upstash-Redact-Fields", header_value(&value)?);
596        }
597    }
598
599    Ok(headers)
600}
601
602fn prefix_forward_headers(headers: &HeaderMap) -> Result<HeaderMap> {
603    let mut prefixed = HeaderMap::new();
604
605    for (name, value) in headers {
606        if should_forward_header(name.as_str()) {
607            let forwarded_name =
608                HeaderName::from_bytes(format!("Upstash-Forward-{}", name.as_str()).as_bytes())
609                    .map_err(|error| Error::InvalidRequest {
610                        message: format!("invalid forwarded header name `{name}`: {error}"),
611                    })?;
612            prefixed.insert(forwarded_name, value.clone());
613        } else {
614            prefixed.insert(name.clone(), value.clone());
615        }
616    }
617
618    Ok(prefixed)
619}
620
621fn should_forward_header(name: &str) -> bool {
622    let lower = name.to_ascii_lowercase();
623    !lower.starts_with("content-type") && !lower.starts_with("upstash-")
624}
625
626fn header_value(value: &str) -> Result<HeaderValue> {
627    HeaderValue::from_str(value).map_err(|error| Error::InvalidRequest {
628        message: format!("invalid header value `{value}`: {error}"),
629    })
630}
qstash_rs/client/request.rs

qstash_rs/client/
request.rs
