pgtask_web/
lib.rs

1#![doc = "Read-only web interface for pgtask."]
2
3mod model;
4mod pages;
5
6use axum::{
7    Router,
8    extract::{Path, Query, State},
9    http::{HeaderMap, HeaderName, StatusCode},
10    response::{Html, IntoResponse, Redirect, Response},
11    routing::{get, post},
12};
13use serde::Deserialize;
14use sqlx::PgPool;
15use thiserror::Error;
16use uuid::Uuid;
17
18use crate::model::{Dashboard, ScheduleDetail, TaskDetail, WorkerDetail};
19
20#[derive(Clone)]
21struct AppState {
22    pool: PgPool,
23    administrator: Option<AdministratorConfig>,
24}
25
26#[derive(Clone, Debug)]
27pub struct AdministratorConfig {
28    pub actor_header: HeaderName,
29}
30
31impl Default for AdministratorConfig {
32    fn default() -> Self {
33        Self {
34            actor_header: HeaderName::from_static("x-pgtask-actor"),
35        }
36    }
37}
38
39#[derive(Debug, Error)]
40enum WebError {
41    #[error("database query failed")]
42    Database(#[from] sqlx::Error),
43    #[error("resource not found")]
44    NotFound,
45    #[error("administrator identity is required")]
46    Unauthorized,
47}
48
49impl IntoResponse for WebError {
50    fn into_response(self) -> Response {
51        let status = match self {
52            Self::Database(_) => StatusCode::INTERNAL_SERVER_ERROR,
53            Self::NotFound => StatusCode::NOT_FOUND,
54            Self::Unauthorized => StatusCode::UNAUTHORIZED,
55        };
56        (status, Html(pages::error(status.as_u16(), &self.to_string()))).into_response()
57    }
58}
59
60#[derive(Deserialize)]
61struct TaskSearch {
62    query: Option<String>,
63}
64
65pub fn application(pool: PgPool) -> Router {
66    build_application(pool, None)
67}
68
69pub fn application_with_administrator(pool: PgPool, config: AdministratorConfig) -> Router {
70    build_application(pool, Some(config))
71}
72
73fn build_application(pool: PgPool, administrator: Option<AdministratorConfig>) -> Router {
74    let mut router = Router::new()
75        .route("/", get(dashboard))
76        .route("/healthz", get(health))
77        .route("/tasks", get(tasks))
78        .route("/tasks/{task_id}", get(task))
79        .route("/schedules", get(schedules))
80        .route("/schedules/{schedule_id}", get(schedule))
81        .route("/workers", get(workers))
82        .route("/workers/{worker_id}", get(worker));
83    if administrator.is_some() {
84        router = router
85            .route("/admin/tasks/{task_id}/cancel", post(cancel_task))
86            .route("/admin/tasks/{task_id}/retry", post(retry_task))
87            .route("/admin/schedules/{schedule_id}/pause", post(pause_schedule))
88            .route("/admin/schedules/{schedule_id}/resume", post(resume_schedule));
89    }
90    router.with_state(AppState { pool, administrator })
91}
92
93async fn health(State(state): State<AppState>) -> Result<&'static str, WebError> {
94    sqlx::query("SELECT 1").execute(&state.pool).await?;
95    Ok("healthy")
96}
97
98async fn dashboard(State(state): State<AppState>) -> Result<Html<String>, WebError> {
99    Ok(Html(pages::dashboard(&Dashboard::load(&state.pool).await?)))
100}
101
102async fn tasks(State(state): State<AppState>, Query(search): Query<TaskSearch>) -> Result<Html<String>, WebError> {
103    Ok(Html(pages::tasks(
104        &model::TaskSummary::search(&state.pool, search.query.as_deref()).await?,
105        search.query.as_deref(),
106    )))
107}
108
109async fn task(State(state): State<AppState>, Path(task_id): Path<Uuid>) -> Result<Html<String>, WebError> {
110    let detail = TaskDetail::load(&state.pool, task_id)
111        .await?
112        .ok_or(WebError::NotFound)?;
113    Ok(Html(pages::task(&detail, state.administrator.is_some())))
114}
115
116async fn schedules(State(state): State<AppState>) -> Result<Html<String>, WebError> {
117    Ok(Html(pages::schedules(&model::ScheduleSummary::all(&state.pool).await?)))
118}
119
120async fn schedule(State(state): State<AppState>, Path(schedule_id): Path<Uuid>) -> Result<Html<String>, WebError> {
121    let detail = ScheduleDetail::load(&state.pool, schedule_id)
122        .await?
123        .ok_or(WebError::NotFound)?;
124    Ok(Html(pages::schedule(&detail, state.administrator.is_some())))
125}
126
127async fn workers(State(state): State<AppState>) -> Result<Html<String>, WebError> {
128    Ok(Html(pages::workers(&model::WorkerSummary::all(&state.pool).await?)))
129}
130
131async fn worker(State(state): State<AppState>, Path(worker_id): Path<Uuid>) -> Result<Html<String>, WebError> {
132    let detail = WorkerDetail::load(&state.pool, worker_id)
133        .await?
134        .ok_or(WebError::NotFound)?;
135    Ok(Html(pages::worker(&detail)))
136}
137
138fn administrator_actor<'a>(state: &AppState, headers: &'a HeaderMap) -> Result<&'a str, WebError> {
139    let config = state.administrator.as_ref().ok_or(WebError::NotFound)?;
140    headers
141        .get(&config.actor_header)
142        .and_then(|value| value.to_str().ok())
143        .filter(|value| !value.is_empty())
144        .ok_or(WebError::Unauthorized)
145}
146
147async fn cancel_task(
148    State(state): State<AppState>,
149    Path(task_id): Path<Uuid>,
150    headers: HeaderMap,
151) -> Result<Redirect, WebError> {
152    let actor = administrator_actor(&state, &headers)?;
153    let changed: bool = sqlx::query_scalar("SELECT pgtask.admin_cancel_task($1, $2)")
154        .bind(task_id)
155        .bind(actor)
156        .fetch_one(&state.pool)
157        .await?;
158    if !changed {
159        return Err(WebError::NotFound);
160    }
161    Ok(Redirect::to(&format!("/tasks/{task_id}")))
162}
163
164async fn retry_task(
165    State(state): State<AppState>,
166    Path(task_id): Path<Uuid>,
167    headers: HeaderMap,
168) -> Result<Redirect, WebError> {
169    let actor = administrator_actor(&state, &headers)?;
170    let changed: bool = sqlx::query_scalar("SELECT pgtask.admin_retry_task($1, $2)")
171        .bind(task_id)
172        .bind(actor)
173        .fetch_one(&state.pool)
174        .await?;
175    if !changed {
176        return Err(WebError::NotFound);
177    }
178    Ok(Redirect::to(&format!("/tasks/{task_id}")))
179}
180
181async fn pause_schedule(
182    State(state): State<AppState>,
183    Path(schedule_id): Path<Uuid>,
184    headers: HeaderMap,
185) -> Result<Redirect, WebError> {
186    set_schedule_paused(&state, schedule_id, &headers, true).await
187}
188
189async fn resume_schedule(
190    State(state): State<AppState>,
191    Path(schedule_id): Path<Uuid>,
192    headers: HeaderMap,
193) -> Result<Redirect, WebError> {
194    set_schedule_paused(&state, schedule_id, &headers, false).await
195}
196
197async fn set_schedule_paused(
198    state: &AppState,
199    schedule_id: Uuid,
200    headers: &HeaderMap,
201    paused: bool,
202) -> Result<Redirect, WebError> {
203    let actor = administrator_actor(state, headers)?;
204    let changed: bool = sqlx::query_scalar("SELECT pgtask.admin_set_schedule_paused($1, $2, $3)")
205        .bind(schedule_id)
206        .bind(paused)
207        .bind(actor)
208        .fetch_one(&state.pool)
209        .await?;
210    if !changed {
211        return Err(WebError::NotFound);
212    }
213    Ok(Redirect::to(&format!("/schedules/{schedule_id}")))
214}
pgtask_web/lib.rs

pgtask_web/
lib.rs
