haste_server/jobs/
operation.rs

1//! The `operation` kind of job: an operation invoked with
2//! `Prefer: respond-async`, run later by the worker exactly as the request
3//! would have run it (the same client, middleware, scopes and access
4//! policies, from the submitter's stored claims), with the resource it
5//! returns served as the job's result.
6
7use crate::error::ServerError;
8use crate::{
9    auth_n::middleware::jwt::User, config::ServerConfig, fhir_client::ServerCTX, jobs::status_url,
10    services::ServerState,
11};
12use haste_fhir_client::{
13    FHIRClient,
14    request::{
15        FHIRAcceptedResponse, FHIRInvokeInstanceRequest, FHIRInvokeSystemRequest,
16        FHIRInvokeTypeRequest, FHIRRequest, FHIRResponse, InvocationRequest, InvokeResponse,
17        Operation,
18    },
19};
20use haste_fhir_model::r4::generated::{
21    resources::{Parameters, ResourceType},
22    terminology::IssueType,
23};
24use haste_fhir_operation_error::OperationOutcomeError;
25use haste_fhir_search::SearchEngine;
26use haste_fhir_terminology::FHIRTerminology;
27use haste_repository::{Repository, async_job::CreateAsyncJob};
28use haste_worker::async_jobs::{HandlerFuture, JobContext, JobHandler, JobOutput};
29use serde::{Deserialize, Serialize};
30use std::sync::Arc;
31
32pub const KIND: &str = "operation";
33
34/// Re-running an operation is not safe in general (a custom one may write),
35/// so an operation job gets one attempt: if its worker dies mid-way, the job
36/// fails rather than runs again.
37const MAX_ATTEMPTS: i32 = 1;
38
39#[derive(Serialize, Deserialize, Clone, Copy, Debug)]
40#[serde(rename_all = "lowercase")]
41enum Level {
42    Instance,
43    Type,
44    System,
45}
46
47/// An invocation as stored in the job's `parameters`: enough to rebuild the
48/// `InvocationRequest` in the worker.
49#[derive(Serialize, Deserialize, Debug)]
50pub struct StoredInvocation {
51    level: Level,
52    #[serde(skip_serializing_if = "Option::is_none")]
53    resource_type: Option<String>,
54    #[serde(skip_serializing_if = "Option::is_none")]
55    id: Option<String>,
56    operation: String,
57    parameters: Parameters,
58}
59
60impl From<&InvocationRequest> for StoredInvocation {
61    fn from(request: &InvocationRequest) -> Self {
62        match request {
63            InvocationRequest::Instance(request) => StoredInvocation {
64                level: Level::Instance,
65                resource_type: Some(request.resource_type.as_ref().to_string()),
66                id: Some(request.id.clone()),
67                operation: request.operation.name().to_string(),
68                parameters: request.parameters.clone(),
69            },
70            InvocationRequest::Type(request) => StoredInvocation {
71                level: Level::Type,
72                resource_type: Some(request.resource_type.as_ref().to_string()),
73                id: None,
74                operation: request.operation.name().to_string(),
75                parameters: request.parameters.clone(),
76            },
77            InvocationRequest::System(request) => StoredInvocation {
78                level: Level::System,
79                resource_type: None,
80                id: None,
81                operation: request.operation.name().to_string(),
82                parameters: request.parameters.clone(),
83            },
84        }
85    }
86}
87
88impl TryFrom<StoredInvocation> for InvocationRequest {
89    type Error = OperationOutcomeError;
90
91    fn try_from(stored: StoredInvocation) -> Result<Self, Self::Error> {
92        let resource_type = |name: Option<String>| {
93            ResourceType::try_from(name.unwrap_or_default()).map_err(|_| {
94                OperationOutcomeError::error(
95                    IssueType::invalid(),
96                    "The stored invocation names an unknown resource type.".to_string(),
97                )
98            })
99        };
100        let operation = Operation::new(&stored.operation);
101
102        Ok(match stored.level {
103            Level::Instance => InvocationRequest::Instance(FHIRInvokeInstanceRequest {
104                operation,
105                resource_type: resource_type(stored.resource_type)?,
106                id: stored.id.unwrap_or_default(),
107                parameters: stored.parameters,
108            }),
109            Level::Type => InvocationRequest::Type(FHIRInvokeTypeRequest {
110                operation,
111                resource_type: resource_type(stored.resource_type)?,
112                parameters: stored.parameters,
113            }),
114            Level::System => InvocationRequest::System(FHIRInvokeSystemRequest {
115                operation,
116                parameters: stored.parameters,
117            }),
118        })
119    }
120}
121
122/// The invocation as a FHIR location, for the job's `request`.
123fn describe(request: &InvocationRequest) -> String {
124    match request {
125        InvocationRequest::Instance(request) => format!(
126            "{}/{}/${}",
127            request.resource_type.as_ref(),
128            request.id,
129            request.operation.name()
130        ),
131        InvocationRequest::Type(request) => {
132            format!(
133                "{}/${}",
134                request.resource_type.as_ref(),
135                request.operation.name()
136            )
137        }
138        InvocationRequest::System(request) => format!("${}", request.operation.name()),
139    }
140}
141
142/// Queues `request` as a job for the caller and answers `202 Accepted` with
143/// its status URL. Called once the operation is known to exist and the
144/// request has passed every check the synchronous form would have.
145pub async fn submit<
146    Repo: Repository + Send + Sync + 'static,
147    Client: FHIRClient<Arc<ServerCTX<Client>>, OperationOutcomeError> + 'static,
148>(
149    repo: &Repo,
150    config: &ServerConfig,
151    ctx: &ServerCTX<Client>,
152    request: &InvocationRequest,
153) -> Result<FHIRResponse, OperationOutcomeError> {
154    let live = repo.count_live_jobs(&ctx.tenant, &ctx.project).await?;
155    if live >= config.jobs.max_live_per_project {
156        return Err(OperationOutcomeError::error(
157            IssueType::throttled(),
158            format!(
159                "Project '{}' already has {live} queued or running job(s); wait for one to \
160                 finish before starting another.",
161                ctx.project.as_ref()
162            ),
163        ));
164    }
165
166    let parameters =
167        serde_json::to_value(StoredInvocation::from(request)).map_err(ServerError::from)?;
168
169    let job = repo
170        .create_job(
171            &ctx.tenant,
172            &ctx.project,
173            CreateAsyncJob {
174                kind: KIND.to_string(),
175                request: describe(request),
176                parameters,
177                claims: ctx.user.claims.clone(),
178                retention: config.jobs.retention(),
179                max_attempts: MAX_ATTEMPTS,
180            },
181        )
182        .await?;
183
184    tracing::info!(
185        job = %job.id,
186        request = %job.request,
187        "Queued an operation as an asynchronous job"
188    );
189
190    let content_location = status_url(&config.api_uri, &ctx.tenant, &ctx.project, &job.id)?;
191
192    Ok(FHIRResponse::Accepted(FHIRAcceptedResponse {
193        content_location: content_location.to_string(),
194    }))
195}
196
197/// Runs `operation` jobs through the server's FHIR client.
198pub struct OperationJobHandler<
199    Repo: Repository + Send + Sync + 'static,
200    Search: SearchEngine + Send + Sync + 'static,
201    Terminology: FHIRTerminology + Send + Sync + 'static,
202> {
203    state: Arc<ServerState<Repo, Search, Terminology>>,
204}
205
206impl<
207    Repo: Repository + Send + Sync + 'static,
208    Search: SearchEngine + Send + Sync + 'static,
209    Terminology: FHIRTerminology + Send + Sync + 'static,
210> OperationJobHandler<Repo, Search, Terminology>
211{
212    #[must_use]
213    pub fn new(state: Arc<ServerState<Repo, Search, Terminology>>) -> Self {
214        Self { state }
215    }
216}
217
218impl<
219    Repo: Repository + Send + Sync + 'static,
220    Search: SearchEngine + Send + Sync + 'static,
221    Terminology: FHIRTerminology + Send + Sync + 'static,
222> JobHandler for OperationJobHandler<Repo, Search, Terminology>
223{
224    fn kind(&self) -> &'static str {
225        KIND
226    }
227
228    fn run<'a>(&'a self, context: &'a JobContext) -> HandlerFuture<'a, JobOutput> {
229        Box::pin(async move {
230            let job = &context.job;
231            let stored: StoredInvocation = serde_json::from_value(job.parameters.0.clone())
232                .map_err(|e| {
233                    OperationOutcomeError::fatal(
234                        IssueType::exception(),
235                        format!("The job's stored invocation could not be read: {e}"),
236                    )
237                })?;
238            let request = InvocationRequest::try_from(stored)?;
239
240            // The submitter's verified claims, so the operation runs with
241            // their scopes and access policies. No token: it was checked at
242            // kick-off, and nothing downstream reads it.
243            let claims = job.claims.0.clone();
244            let ctx = Arc::new(
245                ServerCTX::new(
246                    job.tenant.clone(),
247                    job.project.clone(),
248                    claims.fhir_version.clone(),
249                    Arc::new(User {
250                        token: None,
251                        claims,
252                    }),
253                    self.state.fhir_client.clone(),
254                    self.state.rate_limit.clone(),
255                )
256                .with_tracing_id(Some(format!("job-{}", job.id))),
257            );
258
259            let response = self
260                .state
261                .fhir_client
262                .request(ctx, FHIRRequest::Invocation(request))
263                .await?;
264
265            match response {
266                FHIRResponse::Invoke(InvokeResponse::Instance(response)) => {
267                    Ok(JobOutput::Resource(Box::new(response.resource)))
268                }
269                FHIRResponse::Invoke(InvokeResponse::Type(response)) => {
270                    Ok(JobOutput::Resource(Box::new(response.resource)))
271                }
272                FHIRResponse::Invoke(InvokeResponse::System(response)) => {
273                    Ok(JobOutput::Resource(Box::new(response.resource)))
274                }
275                _ => Err(OperationOutcomeError::fatal(
276                    IssueType::exception(),
277                    "The operation did not return an operation response.".to_string(),
278                )),
279            }
280        })
281    }
282}
haste_server/jobs/operation.rs

haste_server/jobs/
operation.rs
