qstash_rs/client/
logs.rs

1use std::collections::HashMap;
2
3use reqwest::Method;
4use serde::{Deserialize, Serialize};
5
6use super::Client;
7use crate::error::Result;
8
9/// Message state reported by QStash logs.
10#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
12pub enum LogState {
13    /// Message accepted by QStash.
14    Created,
15    /// Message currently being processed.
16    Active,
17    /// Message will be retried.
18    Retry,
19    /// Delivery returned an error and is awaiting retry or final failure.
20    Error,
21    /// Message delivered successfully.
22    Delivered,
23    /// Message exhausted retries.
24    Failed,
25    /// Message was cancelled.
26    Cancelled,
27    /// Message cancellation has been requested.
28    CancelRequested,
29    /// Aggregate filter value used by the API.
30    InProgress,
31    /// Any state not known by this crate yet.
32    #[serde(other)]
33    Unknown,
34}
35
36/// A single QStash log entry.
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38#[serde(rename_all = "camelCase")]
39pub struct LogEntry {
40    /// Timestamp in milliseconds.
41    pub time: u64,
42    /// Message state.
43    pub state: LogState,
44    /// Message identifier.
45    pub message_id: String,
46    /// Next delivery timestamp in milliseconds when one exists.
47    pub next_delivery_time: Option<u64>,
48    /// Delivery error.
49    pub error: Option<String>,
50    /// Destination URL.
51    pub url: Option<String>,
52    /// URL Group name.
53    pub topic_name: Option<String>,
54    /// Endpoint label inside the URL Group.
55    pub endpoint_name: Option<String>,
56    /// Forwarded headers.
57    pub header: Option<HashMap<String, Vec<String>>>,
58    /// UTF-8 body when available.
59    pub body: Option<String>,
60    /// Base64-encoded body for non-UTF-8 payloads.
61    pub body_base64: Option<String>,
62    /// HTTP response status from the destination.
63    pub response_status: Option<u16>,
64    /// User-defined label.
65    pub label: Option<String>,
66    /// Queue name.
67    pub queue_name: Option<String>,
68    /// Schedule identifier.
69    pub schedule_id: Option<String>,
70    /// Caller IP.
71    pub caller_ip: Option<String>,
72    /// Flow control key.
73    pub flow_control_key: Option<String>,
74}
75
76/// Paginated log response.
77#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
78pub struct LogsPage {
79    /// Cursor for the next page.
80    pub cursor: Option<String>,
81    /// Returned log entries.
82    pub logs: Vec<LogEntry>,
83}
84
85/// Query options for listing logs.
86#[derive(Debug, Clone, Default, PartialEq, Eq)]
87pub struct LogFilter {
88    /// Cursor from a previous response.
89    pub cursor: Option<String>,
90    /// Filter by message identifier.
91    pub message_id: Option<String>,
92    /// Filter by message state.
93    pub state: Option<LogState>,
94    /// Filter by destination URL.
95    pub url: Option<String>,
96    /// Filter by URL Group name.
97    pub url_group_name: Option<String>,
98    /// Filter by endpoint label.
99    pub endpoint_name: Option<String>,
100    /// Filter by schedule identifier.
101    pub schedule_id: Option<String>,
102    /// Filter by queue name.
103    pub queue_name: Option<String>,
104    /// Filter by caller IP.
105    pub caller_ip: Option<String>,
106    /// Filter by flow control key.
107    pub flow_control_key: Option<String>,
108    /// Filter by response status.
109    pub response_status: Option<u16>,
110    /// Filter by label.
111    pub label: Option<String>,
112    /// Filter from this inclusive timestamp in milliseconds.
113    pub from_date: Option<u64>,
114    /// Filter to this inclusive timestamp in milliseconds.
115    pub to_date: Option<u64>,
116    /// Maximum number of log entries to return.
117    pub count: Option<u32>,
118}
119
120/// Log operations.
121pub struct LogsApi<'a> {
122    pub(crate) client: &'a Client,
123}
124
125impl LogsApi<'_> {
126    /// Lists published-message logs.
127    pub async fn list(&self, filter: &LogFilter) -> Result<LogsPage> {
128        self.client
129            .http
130            .send_json(
131                Method::GET,
132                "v2/logs",
133                &log_filter_query(filter),
134                None,
135                None,
136            )
137            .await
138    }
139}
140
141fn log_filter_query(filter: &LogFilter) -> Vec<(String, String)> {
142    let mut query = Vec::new();
143
144    if let Some(cursor) = &filter.cursor {
145        query.push((String::from("cursor"), cursor.clone()));
146    }
147    if let Some(message_id) = &filter.message_id {
148        query.push((String::from("messageId"), message_id.clone()));
149    }
150    if let Some(state) = &filter.state {
151        query.push((String::from("state"), serde_state(state)));
152    }
153    if let Some(url) = &filter.url {
154        query.push((String::from("url"), url.clone()));
155    }
156    if let Some(url_group_name) = &filter.url_group_name {
157        query.push((String::from("topicName"), url_group_name.clone()));
158    }
159    if let Some(endpoint_name) = &filter.endpoint_name {
160        query.push((String::from("endpointName"), endpoint_name.clone()));
161    }
162    if let Some(schedule_id) = &filter.schedule_id {
163        query.push((String::from("scheduleId"), schedule_id.clone()));
164    }
165    if let Some(queue_name) = &filter.queue_name {
166        query.push((String::from("queueName"), queue_name.clone()));
167    }
168    if let Some(caller_ip) = &filter.caller_ip {
169        query.push((String::from("callerIp"), caller_ip.clone()));
170    }
171    if let Some(flow_control_key) = &filter.flow_control_key {
172        query.push((String::from("flowControlKey"), flow_control_key.clone()));
173    }
174    if let Some(response_status) = filter.response_status {
175        query.push((String::from("responseStatus"), response_status.to_string()));
176    }
177    if let Some(label) = &filter.label {
178        query.push((String::from("label"), label.clone()));
179    }
180    if let Some(from_date) = filter.from_date {
181        query.push((String::from("fromDate"), from_date.to_string()));
182    }
183    if let Some(to_date) = filter.to_date {
184        query.push((String::from("toDate"), to_date.to_string()));
185    }
186    if let Some(count) = filter.count {
187        query.push((String::from("count"), count.to_string()));
188    }
189
190    query
191}
192
193fn serde_state(state: &LogState) -> String {
194    match state {
195        LogState::Created => String::from("CREATED"),
196        LogState::Active => String::from("ACTIVE"),
197        LogState::Retry => String::from("RETRY"),
198        LogState::Error => String::from("ERROR"),
199        LogState::Delivered => String::from("DELIVERED"),
200        LogState::Failed => String::from("FAILED"),
201        LogState::Cancelled => String::from("CANCELLED"),
202        LogState::CancelRequested => String::from("CANCEL_REQUESTED"),
203        LogState::InProgress => String::from("IN_PROGRESS"),
204        LogState::Unknown => String::from("UNKNOWN"),
205    }
206}
qstash_rs/client/logs.rs

qstash_rs/client/
logs.rs
