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

hammerwork_web/api/
mod.rs
