haste_server/jobs/
mod.rs

1//! The FHIR Asynchronous Request Pattern
2//! (<https://hl7.org/fhir/R4/async.html>, and the Bulk Data IG's version of
3//! it): a request answered with `202 Accepted` and a `Content-Location` to
4//! poll, where the client reads `202` with `X-Progress` and `Retry-After`
5//! while the work runs, then the result, or an error with an
6//! `OperationOutcome`, and can `DELETE` the location to cancel.
7//!
8//! The work itself is an `async_jobs` row run by `haste-health worker`
9//! (see `haste_worker::async_jobs`). This module is the HTTP side: the
10//! status URL, the status and cancel endpoints, and the `Prefer` header
11//! that asks for the pattern. [`operation`] is the first kind of job: an
12//! operation invoked with `Prefer: respond-async`.
13
14use crate::error::ServerError;
15use crate::{auth_n::middleware::jwt::User, route_path::api_v1_path, services::ServerState};
16use axum::{
17    Extension,
18    extract::{Path, State},
19    http::{HeaderMap, HeaderValue, StatusCode, header},
20    response::{IntoResponse, Response},
21};
22use haste_fhir_model::r4::generated::{resources::OperationOutcome, terminology::IssueType};
23use haste_fhir_operation_error::OperationOutcomeError;
24use haste_fhir_search::SearchEngine;
25use haste_fhir_terminology::FHIRTerminology;
26use haste_jwt::{ProjectId, TenantId, UserRole};
27use haste_repository::{
28    Repository,
29    async_job::{AsyncJob, AsyncJobStatus},
30};
31use serde::Deserialize;
32use sqlx::types::time::OffsetDateTime;
33use std::sync::Arc;
34use tower_sessions::cookie::time::{UtcOffset, format_description};
35use url::Url;
36
37pub mod operation;
38
39/// The `Prefer` token that asks for the asynchronous pattern.
40pub const RESPOND_ASYNC: &str = "respond-async";
41
42/// Whether the request's `Prefer` header(s) carry `respond-async`. The header
43/// is a comma-separated list of preferences (`respond-async, return=minimal`)
44/// and may be repeated.
45#[must_use]
46pub fn prefers_async(headers: &HeaderMap) -> bool {
47    headers
48        .get_all("prefer")
49        .iter()
50        .filter_map(|value| value.to_str().ok())
51        .flat_map(|value| value.split(','))
52        .any(|preference| preference.trim().eq_ignore_ascii_case(RESPOND_ASYNC))
53}
54
55/// The absolute URL the client polls for a job: sibling of the project's
56/// FHIR root, so the same token authorizes it, but outside the FHIR routes
57/// since neither its response codes nor its bodies are FHIR interactions.
58pub fn status_url(
59    api_uri: &str,
60    tenant: &TenantId,
61    project: &ProjectId,
62    job_id: &str,
63) -> Result<Url, OperationOutcomeError> {
64    let mut path = api_v1_path(tenant, project);
65    path.push("jobs");
66    path.push(job_id);
67
68    Url::parse(api_uri)
69        .map_err(ServerError::from)?
70        .join(path.to_str().unwrap_or_default())
71        .map_err(|e| ServerError::from(e).into())
72}
73
74#[derive(Deserialize)]
75pub struct JobPath {
76    tenant: TenantId,
77    project: ProjectId,
78    id: String,
79}
80
81/// A job is its submitter's: only they, or a tenant owner or admin, may see
82/// or cancel it. Everyone else in the project is told it does not exist, so
83/// that job ids leak nothing.
84fn owned_by(job: &AsyncJob, user: &User) -> bool {
85    job.submitted_by == user.claims.sub.as_ref()
86        || matches!(user.claims.user_role, UserRole::Owner | UserRole::Admin)
87}
88
89fn not_found(id: &str) -> OperationOutcomeError {
90    OperationOutcomeError::error(IssueType::not_found(), format!("Job '{id}' not found."))
91}
92
93/// Reads a job of the request's project that the caller may see.
94async fn load_job<Repo: Repository + Send + Sync + 'static>(
95    repo: &Repo,
96    path: &JobPath,
97    user: &User,
98) -> Result<AsyncJob, OperationOutcomeError> {
99    let job = repo
100        .read_job(&path.tenant, &path.project, &path.id)
101        .await?
102        .filter(|job| owned_by(job, user))
103        .ok_or_else(|| not_found(&path.id))?;
104
105    Ok(job)
106}
107
108/// An HTTP date (IMF-fixdate), as `Expires` and friends want it.
109fn http_date(at: OffsetDateTime) -> Option<HeaderValue> {
110    let format = format_description::parse_borrowed::<2>(
111        "[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT",
112    )
113    .ok()?;
114
115    at.to_offset(UtcOffset::UTC)
116        .format(&format)
117        .ok()?
118        .parse()
119        .ok()
120}
121
122/// `GET` of the status URL.
123///
124/// - queued or running: `202 Accepted`, `X-Progress`, `Retry-After`, no body;
125/// - completed: `200 OK` and the result, in the content type its handler
126///   chose, with `Expires` saying how long it stays available;
127/// - failed: the `OperationOutcome` the handler produced, with the status
128///   code it implies;
129/// - cancelled, or unknown: `404 Not Found`.
130pub async fn status_handler<
131    Repo: Repository + Send + Sync + 'static,
132    Search: SearchEngine + Send + Sync + 'static,
133    Terminology: FHIRTerminology + Send + Sync + 'static,
134>(
135    Extension(user): Extension<Arc<User>>,
136    Path(path): Path<JobPath>,
137    State(state): State<Arc<ServerState<Repo, Search, Terminology>>>,
138) -> Result<Response, OperationOutcomeError> {
139    let job = load_job(state.repo.as_ref(), &path, &user).await?;
140
141    let mut headers = HeaderMap::new();
142    headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
143
144    match job.status {
145        AsyncJobStatus::Accepted | AsyncJobStatus::InProgress => {
146            let progress = job.progress.as_deref().unwrap_or(job.status.as_str());
147            if let Ok(progress) = HeaderValue::from_str(progress) {
148                headers.insert("x-progress", progress);
149            }
150            headers.insert(
151                header::RETRY_AFTER,
152                HeaderValue::from(state.config.jobs.retry_after_seconds),
153            );
154
155            Ok((StatusCode::ACCEPTED, headers).into_response())
156        }
157        AsyncJobStatus::Completed => {
158            let content_type = job
159                .output_content_type
160                .as_deref()
161                .and_then(|content_type| HeaderValue::from_str(content_type).ok())
162                .unwrap_or_else(|| HeaderValue::from_static("application/json"));
163            headers.insert(header::CONTENT_TYPE, content_type);
164            if let Some(expires) = http_date(job.expires_at) {
165                headers.insert(header::EXPIRES, expires);
166            }
167
168            let body = job
169                .output
170                .map(|output| output.0)
171                .unwrap_or(serde_json::Value::Null);
172            let body = serde_json::to_string(&body).map_err(ServerError::from)?;
173
174            Ok((StatusCode::OK, headers, body).into_response())
175        }
176        AsyncJobStatus::Failed => {
177            let outcome = job
178                .error
179                .and_then(|error| serde_json::from_value::<OperationOutcome>(error.0).ok())
180                .ok_or_else(|| {
181                    OperationOutcomeError::fatal(
182                        IssueType::exception(),
183                        "The job failed.".to_string(),
184                    )
185                })?;
186
187            Err(OperationOutcomeError::new(None, outcome))
188        }
189        AsyncJobStatus::Cancelled => Err(not_found(&path.id)),
190    }
191}
192
193/// `DELETE` of the status URL: cancels the job, or, once it has finished,
194/// gives up its result. `202 Accepted` either way; `404` if there is no such
195/// job or it was already cancelled.
196pub async fn cancel_handler<
197    Repo: Repository + Send + Sync + 'static,
198    Search: SearchEngine + Send + Sync + 'static,
199    Terminology: FHIRTerminology + Send + Sync + 'static,
200>(
201    Extension(user): Extension<Arc<User>>,
202    Path(path): Path<JobPath>,
203    State(state): State<Arc<ServerState<Repo, Search, Terminology>>>,
204) -> Result<Response, OperationOutcomeError> {
205    load_job(state.repo.as_ref(), &path, &user).await?;
206
207    state
208        .repo
209        .cancel_job(&path.tenant, &path.project, &path.id)
210        .await?
211        .ok_or_else(|| not_found(&path.id))?;
212
213    tracing::info!(job = %path.id, "Job cancelled by the client");
214
215    Ok(StatusCode::ACCEPTED.into_response())
216}
haste_server/jobs/mod.rs

haste_server/jobs/
mod.rs
