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    /// Whether `sort_order` asks for ascending order (`asc`, in any case); anything else
250    /// is descending.
251    pub fn is_ascending(&self) -> bool {
252        self.sort_order
253            .as_deref()
254            .is_some_and(|order| order.trim().eq_ignore_ascii_case("asc"))
255    }
256
257    pub fn get_order_by(&self) -> (String, String) {
258        let field = self.sort_by.as_deref().unwrap_or("created_at").to_string();
259        let direction = if self.is_ascending() { "ASC" } else { "DESC" }.to_string();
260        (field, direction)
261    }
262}
263
264/// Common error handling for API endpoints
265pub async fn handle_api_error(
266    err: warp::Rejection,
267) -> Result<impl warp::Reply, std::convert::Infallible> {
268    let response = if err.is_not_found() {
269        ApiResponse::<()>::error("Resource not found".to_string())
270    } else if err
271        .find::<warp::filters::body::BodyDeserializeError>()
272        .is_some()
273    {
274        ApiResponse::<()>::error("Invalid request body".to_string())
275    } else if err.find::<warp::reject::InvalidQuery>().is_some() {
276        ApiResponse::<()>::error("Invalid query parameters".to_string())
277    } else {
278        ApiResponse::<()>::error("Internal server error".to_string())
279    };
280
281    let status = if err.is_not_found() {
282        warp::http::StatusCode::NOT_FOUND
283    } else if err
284        .find::<warp::filters::body::BodyDeserializeError>()
285        .is_some()
286        || err.find::<warp::reject::InvalidQuery>().is_some()
287    {
288        warp::http::StatusCode::BAD_REQUEST
289    } else {
290        warp::http::StatusCode::INTERNAL_SERVER_ERROR
291    };
292
293    Ok(warp::reply::with_status(
294        warp::reply::json(&response),
295        status,
296    ))
297}
298
299/// Extract pagination parameters from query string
300pub fn with_pagination()
301-> impl warp::Filter<Extract = (PaginationParams,), Error = warp::Rejection> + Clone {
302    warp::query::<PaginationParams>()
303}
304
305/// Extract filter parameters from query string  
306pub fn with_filters()
307-> impl warp::Filter<Extract = (FilterParams,), Error = warp::Rejection> + Clone {
308    warp::query::<FilterParams>()
309}
310
311/// Extract sort parameters from query string
312pub fn with_sort() -> impl warp::Filter<Extract = (SortParams,), Error = warp::Rejection> + Clone {
313    warp::query::<SortParams>()
314}
315
316/// Test helpers shared by the handler tests of the API submodules.
317#[cfg(test)]
318pub(crate) mod test_support {
319    use hammerwork::JobQueue;
320    use std::sync::Arc;
321
322    /// A queue whose database is unreachable, so every query fails quickly.
323    pub fn unreachable_queue() -> Arc<JobQueue<sqlx::Postgres>> {
324        let pool = sqlx::postgres::PgPoolOptions::new()
325            .acquire_timeout(std::time::Duration::from_millis(200))
326            .connect_lazy("postgres://nobody:nothing@127.0.0.1:1/none")
327            .expect("lazy pool");
328        Arc::new(JobQueue::new(pool))
329    }
330
331    /// Status code and JSON body of a handler reply.
332    pub async fn body_json(response: warp::reply::Response) -> (u16, serde_json::Value) {
333        use warp::Filter;
334        // `warp::test` needs a filter; hand the response out once via a shared slot.
335        let slot = Arc::new(std::sync::Mutex::new(Some(response)));
336        let filter = warp::any().map(move || slot.lock().unwrap().take().unwrap());
337        let reply = warp::test::request().reply(&filter).await;
338        (
339            reply.status().as_u16(),
340            serde_json::from_slice(reply.body()).unwrap(),
341        )
342    }
343}
344
345#[cfg(test)]
346mod tests {
347    use super::test_support::body_json;
348    use super::*;
349
350    #[test]
351    fn test_api_response_success() {
352        let response = ApiResponse::success("test data");
353        assert!(response.success);
354        assert_eq!(response.data, Some("test data"));
355        assert!(response.error.is_none());
356    }
357
358    #[test]
359    fn test_api_response_error() {
360        let response: ApiResponse<()> = ApiResponse::error("Something went wrong".to_string());
361        assert!(!response.success);
362        assert!(response.data.is_none());
363        assert_eq!(response.error, Some("Something went wrong".to_string()));
364    }
365
366    #[test]
367    fn test_pagination_params_defaults() {
368        let params = PaginationParams::default();
369        assert_eq!(params.get_limit(), 50);
370        assert_eq!(params.get_offset(), 0);
371    }
372
373    #[test]
374    fn test_pagination_params_calculation() {
375        let params = PaginationParams {
376            page: Some(3),
377            limit: Some(20),
378            offset: None,
379        };
380        assert_eq!(params.get_limit(), 20);
381        assert_eq!(params.get_offset(), 40); // (3-1) * 20
382    }
383
384    #[test]
385    fn the_offset_uses_the_clamped_limit_and_never_overflows() {
386        // M13: limit=5000&page=2 serves at most 1000 rows, so page 2 starts at row 1000.
387        let params = PaginationParams {
388            page: Some(2),
389            limit: Some(5000),
390            offset: None,
391        };
392        assert_eq!(params.get_limit(), MAX_PAGE_SIZE);
393        assert_eq!(params.get_offset(), MAX_PAGE_SIZE);
394
395        let huge = PaginationParams {
396            page: Some(u32::MAX),
397            limit: Some(u32::MAX),
398            offset: None,
399        };
400        assert_eq!(
401            huge.get_offset(),
402            u32::MAX,
403            "saturates instead of overflowing"
404        );
405
406        let zero = PaginationParams {
407            page: Some(0),
408            limit: Some(0),
409            offset: None,
410        };
411        assert_eq!(zero.get_limit(), 1, "a zero limit is clamped to one");
412        assert_eq!(zero.get_offset(), 0);
413        let meta = PaginationMeta::new(&zero, 3);
414        assert_eq!(meta.total_pages, 3);
415
416        let explicit = PaginationParams {
417            page: Some(7),
418            limit: Some(10),
419            offset: Some(5),
420        };
421        assert_eq!(explicit.get_offset(), 5, "an explicit offset wins");
422    }
423
424    #[test]
425    fn test_pagination_meta() {
426        let params = PaginationParams {
427            page: Some(2),
428            limit: Some(10),
429            offset: None,
430        };
431        let meta = PaginationMeta::new(&params, 45);
432
433        assert_eq!(meta.page, 2);
434        assert_eq!(meta.limit, 10);
435        assert_eq!(meta.total, 45);
436        assert_eq!(meta.total_pages, 5);
437        assert!(meta.has_next);
438        assert!(meta.has_prev);
439    }
440
441    #[test]
442    fn test_sort_params_defaults() {
443        let params = SortParams {
444            sort_by: None,
445            sort_order: None,
446        };
447        let (field, direction) = params.get_order_by();
448        assert_eq!(field, "created_at");
449        assert_eq!(direction, "DESC");
450    }
451
452    /// #94: `asc` selects ascending order in any case (the jobs list only accepted
453    /// lowercase); everything else is descending.
454    #[test]
455    fn sort_order_is_case_insensitive() {
456        let order = |order: Option<&str>| SortParams {
457            sort_by: Some("created_at".to_string()),
458            sort_order: order.map(str::to_string),
459        };
460        for ascending in ["asc", "ASC", "Asc", " asc "] {
461            assert!(order(Some(ascending)).is_ascending(), "{ascending}");
462            assert_eq!(order(Some(ascending)).get_order_by().1, "ASC");
463        }
464        for descending in [None, Some("desc"), Some("DESC"), Some("ascending")] {
465            assert!(!order(descending).is_ascending(), "{descending:?}");
466            assert_eq!(order(descending).get_order_by().1, "DESC");
467        }
468    }
469
470    #[test]
471    fn test_sort_params_custom() {
472        let params = SortParams {
473            sort_by: Some("name".to_string()),
474            sort_order: Some("asc".to_string()),
475        };
476        let (field, direction) = params.get_order_by();
477        assert_eq!(field, "name");
478        assert_eq!(direction, "ASC");
479    }
480
481    #[tokio::test]
482    async fn test_internal_error_is_500_with_json_error_body() {
483        let (status, body) = body_json(internal_error("Failed to list jobs", &"db down")).await;
484        assert_eq!(status, 500);
485        assert_eq!(body["success"], false);
486        assert_eq!(body["error"], "Failed to list jobs: db down");
487        assert!(body["data"].is_null());
488    }
489
490    #[tokio::test]
491    async fn test_error_reply_uses_given_status() {
492        let (status, body) =
493            body_json(error_reply(warp::http::StatusCode::BAD_REQUEST, "nope")).await;
494        assert_eq!(status, 400);
495        assert_eq!(body["error"], "nope");
496    }
497
498    #[tokio::test]
499    async fn test_json_reply_is_200() {
500        let (status, body) = body_json(json_reply(&ApiResponse::success(5))).await;
501        assert_eq!(status, 200);
502        assert_eq!(body["data"], 5);
503    }
504}
505
506/// #71: the listing endpoints skip rows they cannot decode instead of failing the whole
507/// page with a 500; reading the corrupt job itself still fails. The corrupt rows are
508/// written with raw SQL and removed at the end.
509#[cfg(test)]
510mod undecodable_row_tests {
511    use super::history::JobHistory;
512    use hammerwork::archive::{ArchivalConfig, ArchivalPolicy, ArchivalReason};
513    use hammerwork::{Job, JobId};
514    use serde_json::{Value, json};
515    use std::future::Future;
516    use std::sync::Arc;
517    use warp::Filter;
518
519    /// The ids listed in a paginated response, sorted.
520    fn listed(body: &Value) -> Vec<String> {
521        let mut ids: Vec<String> = body["data"]["items"]
522            .as_array()
523            .unwrap_or_else(|| panic!("no items: {body}"))
524            .iter()
525            .map(|item| item["id"].as_str().unwrap().to_string())
526            .collect();
527        ids.sort();
528        ids
529    }
530
531    /// `sql` runs one statement. `corrupt_archive_id` makes the archived row undecodable
532    /// by giving it an id that is not a UUID (possible on MySQL, where the id is text).
533    async fn listings_skip_undecodable_rows<Q, S, F>(
534        queue: Arc<Q>,
535        sql: S,
536        corrupt_archive_id: bool,
537    ) where
538        Q: JobHistory + 'static,
539        S: Fn(String) -> F,
540        F: Future<Output = ()>,
541    {
542        let routes = super::jobs::routes(queue.clone())
543            .or(super::queues::routes(queue.clone()))
544            .or(super::archive::archive_routes(queue.clone()));
545        let get = |path: String| {
546            let routes = routes.clone();
547            async move {
548                let response = warp::test::request().path(&path).reply(&routes).await;
549                let body: Value = serde_json::from_slice(response.body()).unwrap();
550                (response.status().as_u16(), body)
551            }
552        };
553
554        let tag = uuid::Uuid::new_v4().simple().to_string();
555        let name = format!("undecodable_{tag}");
556        let archive_name = format!("undecodable_archive_{tag}");
557        let enqueue = |queue_name: &str| queue.enqueue(Job::new(queue_name.to_string(), json!({})));
558        let break_timeout = |id: JobId| {
559            sql(format!(
560                "UPDATE hammerwork_jobs SET timeout_seconds = -1 WHERE id = '{id}'"
561            ))
562        };
563
564        let ready = enqueue(&name).await.unwrap();
565        let bad_ready = enqueue(&name).await.unwrap();
566        break_timeout(bad_ready).await;
567        let dead = enqueue(&name).await.unwrap();
568        let bad_dead = enqueue(&name).await.unwrap();
569        for id in [dead, bad_dead] {
570            queue.mark_job_dead(id, "gone").await.unwrap();
571        }
572        break_timeout(bad_dead).await;
573        let mut healthy = vec![ready.to_string(), dead.to_string()];
574        healthy.sort();
575
576        // Job list, queue jobs and search: the healthy jobs, without a 500.
577        for path in [
578            format!("/jobs?queue={name}&limit=100"),
579            format!("/queues/{name}/jobs?limit=100"),
580        ] {
581            let (status, body) = get(path.clone()).await;
582            assert_eq!(status, 200, "{path}: {body}");
583            assert_eq!(listed(&body), healthy, "{path}");
584        }
585        let response = warp::test::request()
586            .method("POST")
587            .path("/jobs/search?limit=100")
588            .json(&json!({"query": "", "queues": [name]}))
589            .reply(&routes)
590            .await;
591        let body: Value = serde_json::from_slice(response.body()).unwrap();
592        assert_eq!(response.status(), 200, "{body}");
593        assert_eq!(listed(&body), healthy, "search");
594
595        // The corrupt job itself still reports the error.
596        let (status, body) = get(format!("/jobs/{bad_ready}")).await;
597        assert_eq!(status, 500, "{body}");
598        assert!(
599            body["error"].as_str().unwrap().contains("timeout_seconds"),
600            "{body}"
601        );
602
603        // Archived jobs
604        let archived = enqueue(&archive_name).await.unwrap();
605        let bad_archived = enqueue(&archive_name).await.unwrap();
606        for id in [archived, bad_archived] {
607            queue.mark_job_dead(id, "gone").await.unwrap();
608        }
609        let policy = ArchivalPolicy::new()
610            .archive_dead_after(chrono::Duration::seconds(0))
611            .enabled(true);
612        queue
613            .archive_jobs(
614                Some(&archive_name),
615                &policy,
616                &ArchivalConfig::new(),
617                ArchivalReason::Manual,
618                None,
619            )
620            .await
621            .unwrap();
622        if corrupt_archive_id {
623            sql(format!(
624                "UPDATE hammerwork_jobs_archive SET id = 'not-a-uuid' WHERE id = '{bad_archived}'"
625            ))
626            .await;
627        }
628        let (status, body) = get(format!("/archive/jobs?queue={archive_name}")).await;
629        assert_eq!(status, 200, "{body}");
630        let ids = listed(&body);
631        assert!(ids.contains(&archived.to_string()), "{body}");
632        if corrupt_archive_id {
633            assert_eq!(ids, [archived.to_string()], "{body}");
634        }
635
636        for (table, queue_name) in [
637            ("hammerwork_jobs", &name),
638            ("hammerwork_jobs_archive", &archive_name),
639        ] {
640            sql(format!(
641                "DELETE FROM {table} WHERE queue_name = '{queue_name}'"
642            ))
643            .await;
644        }
645    }
646
647    #[tokio::test]
648    #[ignore = "requires DATABASE_URL (PostgreSQL)"]
649    async fn postgres_listings_skip_undecodable_rows() {
650        let url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
651        let pool = sqlx::PgPool::connect(&url).await.unwrap();
652        let queue = Arc::new(hammerwork::JobQueue::new(pool.clone()));
653        let sql = |statement: String| {
654            let pool = pool.clone();
655            async move {
656                sqlx::query(&statement).execute(&pool).await.unwrap();
657            }
658        };
659        listings_skip_undecodable_rows(queue, sql, false).await;
660    }
661
662    #[tokio::test]
663    #[ignore = "requires MYSQL_DATABASE_URL"]
664    async fn mysql_listings_skip_undecodable_rows() {
665        let url = std::env::var("MYSQL_DATABASE_URL").expect("MYSQL_DATABASE_URL");
666        let pool = sqlx::MySqlPool::connect(&url).await.unwrap();
667        let queue = Arc::new(hammerwork::JobQueue::new(pool.clone()));
668        let sql = |statement: String| {
669            let pool = pool.clone();
670            async move {
671                sqlx::query(&statement).execute(&pool).await.unwrap();
672            }
673        };
674        listings_skip_undecodable_rows(queue, sql, true).await;
675    }
676}
hammerwork_web/api/mod.rs

hammerwork_web/api/
mod.rs
