hammerwork_web/api/
mod.rs

1//! REST API endpoints for the Hammerwork web dashboard.
2//!
3//! This module provides a comprehensive REST API for job queue management,
4//! including endpoints for jobs, queues, statistics, and system information.
5//! All API responses use a standardized format with proper error handling.
6//!
7//! # API Response Format
8//!
9//! All API endpoints return responses in a consistent format:
10//!
11//! ```rust
12//! use hammerwork_web::api::ApiResponse;
13//! use serde_json::json;
14//!
15//! // Success response
16//! let success_response = ApiResponse::success(json!({"count": 42}));
17//! assert!(success_response.success);
18//! assert!(success_response.data.is_some());
19//! assert!(success_response.error.is_none());
20//!
21//! // Error response
22//! let error_response: ApiResponse<()> = ApiResponse::error("Something went wrong".to_string());
23//! assert!(!error_response.success);
24//! assert!(error_response.data.is_none());
25//! assert!(error_response.error.is_some());
26//! ```
27//!
28//! # Pagination
29//!
30//! Many endpoints support pagination using query parameters:
31//!
32//! ```rust
33//! use hammerwork_web::api::{PaginationParams, PaginationMeta};
34//!
35//! let params = PaginationParams {
36//!     page: Some(2),
37//!     limit: Some(50),
38//!     offset: None,
39//! };
40//!
41//! assert_eq!(params.get_limit(), 50);
42//! assert_eq!(params.get_offset(), 50); // (page-1) * limit
43//!
44//! let meta = PaginationMeta::new(&params, 200); // 200 total items
45//! assert_eq!(meta.page, 2);
46//! assert_eq!(meta.total_pages, 4);
47//! assert!(meta.has_next);
48//! assert!(meta.has_prev);
49//! ```
50
51/// Unwrap a `Result` inside a warp handler returning `Ok(reply)`; on error log it
52/// and return a `500` JSON error response instead of swallowing the failure.
53macro_rules! try_api {
54    ($expr:expr, $context:expr) => {
55        match $expr {
56            Ok(value) => value,
57            Err(e) => return Ok($crate::api::internal_error($context, &e)),
58        }
59    };
60}
61
62pub mod archive;
63pub mod history;
64pub mod jobs;
65pub mod queues;
66pub mod stats;
67pub mod system;
68
69use serde::{Deserialize, Serialize};
70use warp::Reply;
71
72/// Standard API response wrapper
73#[derive(Debug, Serialize)]
74pub struct ApiResponse<T> {
75    pub success: bool,
76    pub data: Option<T>,
77    pub error: Option<String>,
78    pub timestamp: chrono::DateTime<chrono::Utc>,
79}
80
81impl<T> ApiResponse<T> {
82    pub fn success(data: T) -> Self {
83        Self {
84            success: true,
85            data: Some(data),
86            error: None,
87            timestamp: chrono::Utc::now(),
88        }
89    }
90
91    pub fn error(message: String) -> Self {
92        Self {
93            success: false,
94            data: None,
95            error: Some(message),
96            timestamp: chrono::Utc::now(),
97        }
98    }
99}
100
101/// Serialize `body` as a `200 OK` JSON response.
102pub fn json_reply<T: Serialize>(body: &T) -> warp::reply::Response {
103    warp::reply::with_status(warp::reply::json(body), warp::http::StatusCode::OK).into_response()
104}
105
106/// Build a JSON [`ApiResponse`] error body with the given HTTP status.
107pub fn error_reply(
108    status: warp::http::StatusCode,
109    message: impl Into<String>,
110) -> warp::reply::Response {
111    let body = ApiResponse::<()>::error(message.into());
112    warp::reply::with_status(warp::reply::json(&body), status).into_response()
113}
114
115/// Log a failed backend operation and return it as a `500 Internal Server Error`
116/// JSON response (`"{context}: {err}"`).
117///
118/// Handlers use this instead of swallowing the error, so API callers can tell
119/// an outage apart from "no data".
120pub fn internal_error(context: &str, err: &dyn std::fmt::Display) -> warp::reply::Response {
121    tracing::error!(error = %err, "{}", context);
122    error_reply(
123        warp::http::StatusCode::INTERNAL_SERVER_ERROR,
124        format!("{}: {}", context, err),
125    )
126}
127
128/// Turn a failed queue operation into a JSON error reply: a missing job is a `404`, a
129/// transition the job's status does not allow a `409`, anything else a logged `500`.
130pub fn queue_error_reply(
131    context: &str,
132    err: &hammerwork::HammerworkError,
133) -> warp::reply::Response {
134    use hammerwork::HammerworkError;
135    match err {
136        HammerworkError::JobNotFound { .. } => error_reply(
137            warp::http::StatusCode::NOT_FOUND,
138            format!("{}: {}", context, err),
139        ),
140        HammerworkError::InvalidJobTransition { .. } => error_reply(
141            warp::http::StatusCode::CONFLICT,
142            format!("{}: {}", context, err),
143        ),
144        _ => internal_error(context, err),
145    }
146}
147
148/// Pagination parameters
149#[derive(Debug, Deserialize)]
150pub struct PaginationParams {
151    pub page: Option<u32>,
152    pub limit: Option<u32>,
153    pub offset: Option<u32>,
154}
155
156impl Default for PaginationParams {
157    fn default() -> Self {
158        Self {
159            page: Some(1),
160            limit: Some(50),
161            offset: None,
162        }
163    }
164}
165
166/// The page size when a request does not give one.
167pub const DEFAULT_PAGE_SIZE: u32 = 50;
168
169/// The largest page size a request may ask for; larger limits are clamped to it.
170pub const MAX_PAGE_SIZE: u32 = 1000;
171
172impl PaginationParams {
173    /// The offset of the first item: `offset` if given, else `(page - 1) * limit` computed
174    /// with the clamped [`get_limit`](Self::get_limit), saturating instead of overflowing.
175    pub fn get_offset(&self) -> u32 {
176        if let Some(offset) = self.offset {
177            offset
178        } else {
179            let page = self.page.unwrap_or(1);
180            page.saturating_sub(1).saturating_mul(self.get_limit())
181        }
182    }
183
184    /// The page size: `limit`, clamped to `1..=`[`MAX_PAGE_SIZE`].
185    pub fn get_limit(&self) -> u32 {
186        self.limit
187            .unwrap_or(DEFAULT_PAGE_SIZE)
188            .clamp(1, MAX_PAGE_SIZE)
189    }
190}
191
192/// Pagination metadata for responses
193#[derive(Debug, Serialize)]
194pub struct PaginationMeta {
195    pub page: u32,
196    pub limit: u32,
197    pub offset: u32,
198    pub total: u64,
199    pub total_pages: u32,
200    pub has_next: bool,
201    pub has_prev: bool,
202}
203
204impl PaginationMeta {
205    pub fn new(params: &PaginationParams, total: u64) -> Self {
206        let limit = params.get_limit();
207        let offset = params.get_offset();
208        let page = params.page.unwrap_or(1);
209        let total_pages = u32::try_from(total.div_ceil(u64::from(limit))).unwrap_or(u32::MAX);
210
211        Self {
212            page,
213            limit,
214            offset,
215            total,
216            total_pages,
217            has_next: page < total_pages,
218            has_prev: page > 1,
219        }
220    }
221}
222
223/// Paginated response wrapper
224#[derive(Debug, Serialize)]
225pub struct PaginatedResponse<T> {
226    pub items: Vec<T>,
227    pub pagination: PaginationMeta,
228}
229
230/// Query filter parameters
231#[derive(Debug, Deserialize, Default)]
232pub struct FilterParams {
233    pub status: Option<String>,
234    pub priority: Option<String>,
235    pub queue: Option<String>,
236    pub created_after: Option<chrono::DateTime<chrono::Utc>>,
237    pub created_before: Option<chrono::DateTime<chrono::Utc>>,
238    pub search: Option<String>,
239}
240
241/// Sort parameters
242#[derive(Debug, Deserialize, Default)]
243pub struct SortParams {
244    pub sort_by: Option<String>,
245    pub sort_order: Option<String>,
246}
247
248impl SortParams {
249    pub fn get_order_by(&self) -> (String, String) {
250        let field = self.sort_by.as_deref().unwrap_or("created_at").to_string();
251        let direction = match self.sort_order.as_deref() {
252            Some("asc") | Some("ASC") => "ASC".to_string(),
253            _ => "DESC".to_string(),
254        };
255        (field, direction)
256    }
257}
258
259/// Common error handling for API endpoints
260pub async fn handle_api_error(
261    err: warp::Rejection,
262) -> Result<impl warp::Reply, std::convert::Infallible> {
263    let response = if err.is_not_found() {
264        ApiResponse::<()>::error("Resource not found".to_string())
265    } else if err
266        .find::<warp::filters::body::BodyDeserializeError>()
267        .is_some()
268    {
269        ApiResponse::<()>::error("Invalid request body".to_string())
270    } else if err.find::<warp::reject::InvalidQuery>().is_some() {
271        ApiResponse::<()>::error("Invalid query parameters".to_string())
272    } else {
273        ApiResponse::<()>::error("Internal server error".to_string())
274    };
275
276    let status = if err.is_not_found() {
277        warp::http::StatusCode::NOT_FOUND
278    } else if err
279        .find::<warp::filters::body::BodyDeserializeError>()
280        .is_some()
281        || err.find::<warp::reject::InvalidQuery>().is_some()
282    {
283        warp::http::StatusCode::BAD_REQUEST
284    } else {
285        warp::http::StatusCode::INTERNAL_SERVER_ERROR
286    };
287
288    Ok(warp::reply::with_status(
289        warp::reply::json(&response),
290        status,
291    ))
292}
293
294/// Extract pagination parameters from query string
295pub fn with_pagination()
296-> impl warp::Filter<Extract = (PaginationParams,), Error = warp::Rejection> + Clone {
297    warp::query::<PaginationParams>()
298}
299
300/// Extract filter parameters from query string  
301pub fn with_filters()
302-> impl warp::Filter<Extract = (FilterParams,), Error = warp::Rejection> + Clone {
303    warp::query::<FilterParams>()
304}
305
306/// Extract sort parameters from query string
307pub fn with_sort() -> impl warp::Filter<Extract = (SortParams,), Error = warp::Rejection> + Clone {
308    warp::query::<SortParams>()
309}
310
311/// Test helpers shared by the handler tests of the API submodules.
312#[cfg(test)]
313pub(crate) mod test_support {
314    use hammerwork::JobQueue;
315    use std::sync::Arc;
316
317    /// A queue whose database is unreachable, so every query fails quickly.
318    pub fn unreachable_queue() -> Arc<JobQueue<sqlx::Postgres>> {
319        let pool = sqlx::postgres::PgPoolOptions::new()
320            .acquire_timeout(std::time::Duration::from_millis(200))
321            .connect_lazy("postgres://nobody:nothing@127.0.0.1:1/none")
322            .expect("lazy pool");
323        Arc::new(JobQueue::new(pool))
324    }
325
326    /// Status code and JSON body of a handler reply.
327    pub async fn body_json(response: warp::reply::Response) -> (u16, serde_json::Value) {
328        use warp::Filter;
329        // `warp::test` needs a filter; hand the response out once via a shared slot.
330        let slot = Arc::new(std::sync::Mutex::new(Some(response)));
331        let filter = warp::any().map(move || slot.lock().unwrap().take().unwrap());
332        let reply = warp::test::request().reply(&filter).await;
333        (
334            reply.status().as_u16(),
335            serde_json::from_slice(reply.body()).unwrap(),
336        )
337    }
338}
339
340#[cfg(test)]
341mod tests {
342    use super::test_support::body_json;
343    use super::*;
344
345    #[test]
346    fn test_api_response_success() {
347        let response = ApiResponse::success("test data");
348        assert!(response.success);
349        assert_eq!(response.data, Some("test data"));
350        assert!(response.error.is_none());
351    }
352
353    #[test]
354    fn test_api_response_error() {
355        let response: ApiResponse<()> = ApiResponse::error("Something went wrong".to_string());
356        assert!(!response.success);
357        assert!(response.data.is_none());
358        assert_eq!(response.error, Some("Something went wrong".to_string()));
359    }
360
361    #[test]
362    fn test_pagination_params_defaults() {
363        let params = PaginationParams::default();
364        assert_eq!(params.get_limit(), 50);
365        assert_eq!(params.get_offset(), 0);
366    }
367
368    #[test]
369    fn test_pagination_params_calculation() {
370        let params = PaginationParams {
371            page: Some(3),
372            limit: Some(20),
373            offset: None,
374        };
375        assert_eq!(params.get_limit(), 20);
376        assert_eq!(params.get_offset(), 40); // (3-1) * 20
377    }
378
379    #[test]
380    fn the_offset_uses_the_clamped_limit_and_never_overflows() {
381        // M13: limit=5000&page=2 serves at most 1000 rows, so page 2 starts at row 1000.
382        let params = PaginationParams {
383            page: Some(2),
384            limit: Some(5000),
385            offset: None,
386        };
387        assert_eq!(params.get_limit(), MAX_PAGE_SIZE);
388        assert_eq!(params.get_offset(), MAX_PAGE_SIZE);
389
390        let huge = PaginationParams {
391            page: Some(u32::MAX),
392            limit: Some(u32::MAX),
393            offset: None,
394        };
395        assert_eq!(
396            huge.get_offset(),
397            u32::MAX,
398            "saturates instead of overflowing"
399        );
400
401        let zero = PaginationParams {
402            page: Some(0),
403            limit: Some(0),
404            offset: None,
405        };
406        assert_eq!(zero.get_limit(), 1, "a zero limit is clamped to one");
407        assert_eq!(zero.get_offset(), 0);
408        let meta = PaginationMeta::new(&zero, 3);
409        assert_eq!(meta.total_pages, 3);
410
411        let explicit = PaginationParams {
412            page: Some(7),
413            limit: Some(10),
414            offset: Some(5),
415        };
416        assert_eq!(explicit.get_offset(), 5, "an explicit offset wins");
417    }
418
419    #[test]
420    fn test_pagination_meta() {
421        let params = PaginationParams {
422            page: Some(2),
423            limit: Some(10),
424            offset: None,
425        };
426        let meta = PaginationMeta::new(&params, 45);
427
428        assert_eq!(meta.page, 2);
429        assert_eq!(meta.limit, 10);
430        assert_eq!(meta.total, 45);
431        assert_eq!(meta.total_pages, 5);
432        assert!(meta.has_next);
433        assert!(meta.has_prev);
434    }
435
436    #[test]
437    fn test_sort_params_defaults() {
438        let params = SortParams {
439            sort_by: None,
440            sort_order: None,
441        };
442        let (field, direction) = params.get_order_by();
443        assert_eq!(field, "created_at");
444        assert_eq!(direction, "DESC");
445    }
446
447    #[test]
448    fn test_sort_params_custom() {
449        let params = SortParams {
450            sort_by: Some("name".to_string()),
451            sort_order: Some("asc".to_string()),
452        };
453        let (field, direction) = params.get_order_by();
454        assert_eq!(field, "name");
455        assert_eq!(direction, "ASC");
456    }
457
458    #[tokio::test]
459    async fn test_internal_error_is_500_with_json_error_body() {
460        let (status, body) = body_json(internal_error("Failed to list jobs", &"db down")).await;
461        assert_eq!(status, 500);
462        assert_eq!(body["success"], false);
463        assert_eq!(body["error"], "Failed to list jobs: db down");
464        assert!(body["data"].is_null());
465    }
466
467    #[tokio::test]
468    async fn test_error_reply_uses_given_status() {
469        let (status, body) =
470            body_json(error_reply(warp::http::StatusCode::BAD_REQUEST, "nope")).await;
471        assert_eq!(status, 400);
472        assert_eq!(body["error"], "nope");
473    }
474
475    #[tokio::test]
476    async fn test_json_reply_is_200() {
477        let (status, body) = body_json(json_reply(&ApiResponse::success(5))).await;
478        assert_eq!(status, 200);
479        assert_eq!(body["data"], 5);
480    }
481}
482
483/// #71: the listing endpoints skip rows they cannot decode instead of failing the whole
484/// page with a 500; reading the corrupt job itself still fails. The corrupt rows are
485/// written with raw SQL and removed at the end.
486#[cfg(test)]
487mod undecodable_row_tests {
488    use super::history::JobHistory;
489    use hammerwork::archive::{ArchivalConfig, ArchivalPolicy, ArchivalReason};
490    use hammerwork::{Job, JobId};
491    use serde_json::{Value, json};
492    use std::future::Future;
493    use std::sync::Arc;
494    use warp::Filter;
495
496    /// The ids listed in a paginated response, sorted.
497    fn listed(body: &Value) -> Vec<String> {
498        let mut ids: Vec<String> = body["data"]["items"]
499            .as_array()
500            .unwrap_or_else(|| panic!("no items: {body}"))
501            .iter()
502            .map(|item| item["id"].as_str().unwrap().to_string())
503            .collect();
504        ids.sort();
505        ids
506    }
507
508    /// `sql` runs one statement. `corrupt_archive_id` makes the archived row undecodable
509    /// by giving it an id that is not a UUID (possible on MySQL, where the id is text).
510    async fn listings_skip_undecodable_rows<Q, S, F>(
511        queue: Arc<Q>,
512        sql: S,
513        corrupt_archive_id: bool,
514    ) where
515        Q: JobHistory + 'static,
516        S: Fn(String) -> F,
517        F: Future<Output = ()>,
518    {
519        let routes = super::jobs::routes(queue.clone())
520            .or(super::queues::routes(queue.clone()))
521            .or(super::archive::archive_routes(queue.clone()));
522        let get = |path: String| {
523            let routes = routes.clone();
524            async move {
525                let response = warp::test::request().path(&path).reply(&routes).await;
526                let body: Value = serde_json::from_slice(response.body()).unwrap();
527                (response.status().as_u16(), body)
528            }
529        };
530
531        let tag = uuid::Uuid::new_v4().simple().to_string();
532        let name = format!("undecodable_{tag}");
533        let archive_name = format!("undecodable_archive_{tag}");
534        let enqueue = |queue_name: &str| queue.enqueue(Job::new(queue_name.to_string(), json!({})));
535        let break_timeout = |id: JobId| {
536            sql(format!(
537                "UPDATE hammerwork_jobs SET timeout_seconds = -1 WHERE id = '{id}'"
538            ))
539        };
540
541        let ready = enqueue(&name).await.unwrap();
542        let bad_ready = enqueue(&name).await.unwrap();
543        break_timeout(bad_ready).await;
544        let dead = enqueue(&name).await.unwrap();
545        let bad_dead = enqueue(&name).await.unwrap();
546        for id in [dead, bad_dead] {
547            queue.mark_job_dead(id, "gone").await.unwrap();
548        }
549        break_timeout(bad_dead).await;
550        let mut healthy = vec![ready.to_string(), dead.to_string()];
551        healthy.sort();
552
553        // Job list, queue jobs and search: the healthy jobs, without a 500.
554        for path in [
555            format!("/jobs?queue={name}&limit=100"),
556            format!("/queues/{name}/jobs?limit=100"),
557        ] {
558            let (status, body) = get(path.clone()).await;
559            assert_eq!(status, 200, "{path}: {body}");
560            assert_eq!(listed(&body), healthy, "{path}");
561        }
562        let response = warp::test::request()
563            .method("POST")
564            .path("/jobs/search?limit=100")
565            .json(&json!({"query": "", "queues": [name]}))
566            .reply(&routes)
567            .await;
568        let body: Value = serde_json::from_slice(response.body()).unwrap();
569        assert_eq!(response.status(), 200, "{body}");
570        assert_eq!(listed(&body), healthy, "search");
571
572        // The corrupt job itself still reports the error.
573        let (status, body) = get(format!("/jobs/{bad_ready}")).await;
574        assert_eq!(status, 500, "{body}");
575        assert!(
576            body["error"].as_str().unwrap().contains("timeout_seconds"),
577            "{body}"
578        );
579
580        // Archived jobs
581        let archived = enqueue(&archive_name).await.unwrap();
582        let bad_archived = enqueue(&archive_name).await.unwrap();
583        for id in [archived, bad_archived] {
584            queue.mark_job_dead(id, "gone").await.unwrap();
585        }
586        let policy = ArchivalPolicy::new()
587            .archive_dead_after(chrono::Duration::seconds(0))
588            .enabled(true);
589        queue
590            .archive_jobs(
591                Some(&archive_name),
592                &policy,
593                &ArchivalConfig::new(),
594                ArchivalReason::Manual,
595                None,
596            )
597            .await
598            .unwrap();
599        if corrupt_archive_id {
600            sql(format!(
601                "UPDATE hammerwork_jobs_archive SET id = 'not-a-uuid' WHERE id = '{bad_archived}'"
602            ))
603            .await;
604        }
605        let (status, body) = get(format!("/archive/jobs?queue={archive_name}")).await;
606        assert_eq!(status, 200, "{body}");
607        let ids = listed(&body);
608        assert!(ids.contains(&archived.to_string()), "{body}");
609        if corrupt_archive_id {
610            assert_eq!(ids, [archived.to_string()], "{body}");
611        }
612
613        for (table, queue_name) in [
614            ("hammerwork_jobs", &name),
615            ("hammerwork_jobs_archive", &archive_name),
616        ] {
617            sql(format!(
618                "DELETE FROM {table} WHERE queue_name = '{queue_name}'"
619            ))
620            .await;
621        }
622    }
623
624    #[tokio::test]
625    #[ignore = "requires DATABASE_URL (PostgreSQL)"]
626    async fn postgres_listings_skip_undecodable_rows() {
627        let url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
628        let pool = sqlx::PgPool::connect(&url).await.unwrap();
629        let queue = Arc::new(hammerwork::JobQueue::new(pool.clone()));
630        let sql = |statement: String| {
631            let pool = pool.clone();
632            async move {
633                sqlx::query(&statement).execute(&pool).await.unwrap();
634            }
635        };
636        listings_skip_undecodable_rows(queue, sql, false).await;
637    }
638
639    #[tokio::test]
640    #[ignore = "requires MYSQL_DATABASE_URL"]
641    async fn mysql_listings_skip_undecodable_rows() {
642        let url = std::env::var("MYSQL_DATABASE_URL").expect("MYSQL_DATABASE_URL");
643        let pool = sqlx::MySqlPool::connect(&url).await.unwrap();
644        let queue = Arc::new(hammerwork::JobQueue::new(pool.clone()));
645        let sql = |statement: String| {
646            let pool = pool.clone();
647            async move {
648                sqlx::query(&statement).execute(&pool).await.unwrap();
649            }
650        };
651        listings_skip_undecodable_rows(queue, sql, true).await;
652    }
653}
hammerwork_web/api/mod.rs

hammerwork_web/api/
mod.rs
