ferrobridge_term/client.rs
1// SPDX-FileCopyrightText: Cadasto B.V.
2// SPDX-License-Identifier: BUSL-1.1
3
4//! The client itself: one configured terminology server, one request at a
5//! time.
6
7use crate::config::{Config, Credentials, TokenFailure};
8use crate::error::{Error, UpstreamError};
9use crate::outcome::{
10 BatchOutcome, Coded, LookupOutcome, Request, TranslateOutcome, Translation, ValidateOutcome,
11 Validation,
12};
13use backon::{ExponentialBuilder, Retryable};
14use http::header::{ACCEPT, CONTENT_TYPE, WWW_AUTHENTICATE};
15use http::{HeaderMap, Method, StatusCode};
16use secrecy::ExposeSecret;
17use std::sync::Arc;
18use url::Url;
19
20/// The FHIR JSON media type (<https://hl7.org/fhir/R4/http.html#mime-type>).
21pub const FHIR_JSON: &str = "application/fhir+json";
22
23/// The path of the capability statement a FHIR server publishes
24/// (<https://hl7.org/fhir/R4/http.html#capabilities>).
25const METADATA: &str = "metadata";
26
27/// One request, described so that every attempt builds the same thing.
28struct Call {
29 /// The operation, as `Resource/$code`, or the batch marker.
30 operation: &'static str,
31 /// The absolute URL.
32 url: Url,
33 /// The FHIR JSON request body.
34 body: String,
35}
36
37/// What the server answered, read whole.
38pub(crate) struct Answer {
39 /// The URL that was called.
40 pub(crate) url: Url,
41 /// The status the server answered with.
42 pub(crate) status: StatusCode,
43 /// The response body, as received.
44 pub(crate) body: String,
45}
46
47/// A client for one FHIR terminology server.
48///
49/// The client is cheap to clone: the HTTP stack and the configuration are
50/// shared, so a clone carries the same connection pool.
51#[derive(Debug, Clone)]
52pub struct Client {
53 /// The HTTP stack, with the configured timeout.
54 http: reqwest::Client,
55 /// The server this client talks to.
56 config: Arc<Config>,
57}
58
59impl Client {
60 /// Returns a client for the server `config` describes.
61 ///
62 /// This constructor is the only way to obtain a client, so a deployment
63 /// that configures no terminology server holds no [`Client`] at all and
64 /// the absence is a compile-time `Option<Client>` rather than a runtime
65 /// fallback.
66 ///
67 /// # Errors
68 /// Returns [`Error::BaseUrl`] when the base URL cannot carry a path, and
69 /// [`Error::ClientBuild`] when the HTTP stack refuses the configuration.
70 pub fn new(config: Config) -> Result<Self, Error> {
71 if config.base_url.cannot_be_a_base() {
72 return Err(Error::BaseUrl {
73 base: config.base_url.clone(),
74 });
75 }
76 let http = reqwest::Client::builder()
77 .timeout(config.timeout)
78 .build()
79 .map_err(|source| Error::ClientBuild { source })?;
80 Ok(Self {
81 http,
82 config: Arc::new(config),
83 })
84 }
85
86 /// Returns the configuration this client was built from.
87 #[must_use]
88 pub fn config(&self) -> &Config {
89 &self.config
90 }
91
92 /// Checks that the configured release can carry `request`, sending
93 /// nothing.
94 ///
95 /// A caller that sends several requests uses this to set aside one the
96 /// release cannot carry, since a batch holding it is refused whole.
97 ///
98 /// # Errors
99 /// Returns [`Error::UnsupportedParameter`] when `request` names a
100 /// parameter the configured release does not declare: a concept map
101 /// `url` or a value set version on an STU3 client.
102 pub fn check(&self, request: &Request) -> Result<(), Error> {
103 let version = self.config.wire_version;
104 match crate::wire::unsupported(version, request) {
105 Some(parameter) => Err(Error::UnsupportedParameter {
106 operation: request.operation(),
107 parameter,
108 wire_version: version,
109 }),
110 None => Ok(()),
111 }
112 }
113
114 /// Returns the status the server answers `GET [base]/metadata` with.
115 ///
116 /// Every FHIR REST server states its capabilities at that path
117 /// (<https://hl7.org/fhir/R4/http.html#capabilities>), so a reachability
118 /// question needs no terminology operation. The call is not retried, and
119 /// every status is returned as it stands, `401` and `404` included; under
120 /// [`Credentials::OAuth`] a refused token is renewed once first.
121 ///
122 /// # Errors
123 /// Returns [`Error::BaseUrl`] when the base URL cannot carry a path,
124 /// [`Error::Token`] when no token could be obtained, [`Error::Timeout`]
125 /// when the server did not answer inside the configured budget, and
126 /// [`Error::Transport`] when the request never reached it.
127 pub async fn reachability(&self) -> Result<StatusCode, Error> {
128 let url = self.url(METADATA)?;
129 let (response, renewal) = self
130 .send(&url, || {
131 self.http
132 .request(Method::GET, url.clone())
133 .header(ACCEPT, FHIR_JSON)
134 })
135 .await?;
136 if let Some(source) = renewal {
137 return Err(Error::Token { url, source });
138 }
139 let status = response.status();
140 tracing::debug!(path = url.path(), status = %status, "the terminology server answered a probe");
141 Ok(status)
142 }
143
144 /// Returns the response to the request `build` makes under the configured
145 /// credentials, with the provider's failure when the server refused an
146 /// OAuth token and no new one could be obtained.
147 ///
148 /// A refused OAuth token is dropped and the request sent once more under a
149 /// new one; a second `401` is returned as it stands.
150 async fn send(
151 &self,
152 url: &Url,
153 build: impl Fn() -> reqwest::RequestBuilder,
154 ) -> Result<(reqwest::Response, Option<TokenFailure>), Error> {
155 let provider = match self.config.credentials.as_ref() {
156 None => return Ok((dispatch(url, build()).await?, None)),
157 Some(Credentials::Basic { user, password }) => {
158 let request = build().basic_auth(user, Some(password.expose_secret()));
159 return Ok((dispatch(url, request).await?, None));
160 }
161 Some(Credentials::OAuth(provider)) => provider,
162 };
163 let token = provider.token().await.map_err(|source| Error::Token {
164 url: url.clone(),
165 source,
166 })?;
167 let response = dispatch(url, build().bearer_auth(token.expose_secret())).await?;
168 // NOTE: RFC 9110 ยง15.5.2, a `401` means the request was not applied, so
169 // sending it once more under a renewed token cannot apply it twice.
170 if response.status() != StatusCode::UNAUTHORIZED {
171 return Ok((response, None));
172 }
173 provider.invalidate(&token);
174 match provider.token().await {
175 Ok(renewed) => {
176 tracing::debug!(
177 path = url.path(),
178 "the terminology server refused the token; re-sending once under a renewed one"
179 );
180 let request = build().bearer_auth(renewed.expose_secret());
181 Ok((dispatch(url, request).await?, None))
182 }
183 Err(source) => Ok((response, Some(source))),
184 }
185 }
186
187 /// Resolves the display of `code` in `system`.
188 ///
189 /// A server that does not know the code answers [`LookupOutcome::NotFound`]
190 /// carrying its `OperationOutcome`; the display is never an empty string.
191 ///
192 /// # Errors
193 /// Returns [`Error::Refused`] when the server refused the operation for
194 /// another reason, and [`Error`] otherwise when the call did not reach an
195 /// answer or the answer could not be decoded.
196 pub async fn lookup(
197 &self,
198 system: &str,
199 code: &str,
200 version: Option<&str>,
201 ) -> Result<LookupOutcome, Error> {
202 let request = Request::Lookup {
203 system: system.to_owned(),
204 code: code.to_owned(),
205 version: version.map(str::to_owned),
206 };
207 let answer = self.operation(&request).await?;
208 crate::decode::lookup(self.config.wire_version, &answer)
209 }
210
211 /// Translates the code `translation` names, through the concept map,
212 /// into the target system and in the direction it names.
213 ///
214 /// `url`, `targetsystem` and `reverse` are each sent only when the
215 /// translation sets them, as R4 declares each `0..1`
216 /// (<https://hl7.org/fhir/R4/conceptmap-operation-translate.html>).
217 /// Every match the server returned is carried with its equivalence, so the
218 /// caller decides which ones count;
219 /// [`TranslateOutcome::accepted`] yields the `equivalent` and `equal` ones.
220 ///
221 /// # Examples
222 ///
223 /// A deployment with no terminology server holds no client, and a
224 /// translation it cannot make is a refusal:
225 ///
226 /// ```rust,no_run
227 /// use ferrobridge_term::client::Client;
228 /// use ferrobridge_term::outcome::Translation;
229 ///
230 /// # fn main() {
231 /// # let terminology: Option<Client> = None;
232 /// # let _call = async move {
233 /// let Some(client) = terminology.as_ref() else {
234 /// return Err("no terminology server is configured".into());
235 /// };
236 /// let translation = Translation::new("http://example.org/source", "alpha")
237 /// .with_concept_map("http://example.org/ConceptMap/source-to-target");
238 /// let outcome = client.translate(&translation).await?;
239 /// let Some(translated) = outcome.accepted().next() else {
240 /// return Err("no equivalent translation".into());
241 /// };
242 /// let code = translated.concept.as_ref().and_then(|concept| concept.code.as_deref());
243 /// assert!(code.is_some(), "an accepted match names its target code");
244 /// # Ok::<(), Box<dyn std::error::Error>>(())
245 /// # };
246 /// # }
247 /// ```
248 ///
249 /// # Errors
250 /// Returns [`Error::Refused`] when the server refused the operation, for
251 /// example because it serves no such concept map, and [`Error`] otherwise
252 /// when the call did not reach an answer or the answer could not be
253 /// decoded. A refusal is never a missing translation. Returns
254 /// [`Error::UnsupportedParameter`] when the translation names a concept
255 /// map on an STU3 client, whose `$translate` has no `url`; nothing is
256 /// sent.
257 pub async fn translate(&self, translation: &Translation) -> Result<TranslateOutcome, Error> {
258 let request = Request::Translate(translation.clone());
259 let answer = self.operation(&request).await?;
260 crate::decode::translate(self.config.wire_version, &answer)
261 }
262
263 /// Validates that `code` from `system` is a member of `value_set_url`.
264 ///
265 /// # Errors
266 /// Returns [`Error::Refused`] when the server refused the operation, for
267 /// example because it serves no such value set, and [`Error`] otherwise
268 /// when the call did not reach an answer or the answer could not be
269 /// decoded. A refusal is never an invalid code.
270 pub async fn validate_code(
271 &self,
272 value_set_url: &str,
273 system: &str,
274 code: &str,
275 ) -> Result<ValidateOutcome, Error> {
276 self.validate(&Validation {
277 value_set_url: value_set_url.to_owned(),
278 value_set_version: None,
279 coded: Coded::Code {
280 system: system.to_owned(),
281 code: code.to_owned(),
282 version: None,
283 display: None,
284 },
285 })
286 .await
287 }
288
289 /// Validates that the coded value of `validation` is a member of its
290 /// value set, in any of the three forms the operation accepts: a code
291 /// with its system, version and display, a `Coding`, or a
292 /// `CodeableConcept`
293 /// (<https://hl7.org/fhir/R4/valueset-operation-validate-code.html>).
294 ///
295 /// # Errors
296 /// Returns [`Error::Refused`] when the server refused the operation, and
297 /// [`Error`] otherwise when the call did not reach an answer or the answer
298 /// could not be decoded. A refusal is never an invalid code. Returns
299 /// [`Error::UnsupportedParameter`] when the validation names a value set
300 /// version on an STU3 client, whose `$validate-code` has no
301 /// `valueSetVersion`; nothing is sent.
302 pub async fn validate(&self, validation: &Validation) -> Result<ValidateOutcome, Error> {
303 let request = Request::ValidateCode(validation.clone());
304 let answer = self.operation(&request).await?;
305 crate::decode::validate_code(self.config.wire_version, &answer)
306 }
307
308 /// Validates every one of `validations`, answering them by position: one
309 /// call for one validation, one `batch` `Bundle` for several.
310 ///
311 /// An entry the server refused is that entry's own typed error and leaves
312 /// the others alone, as in [`Client::batch`].
313 ///
314 /// # Errors
315 /// Returns [`Error::BatchArity`] when a batch answer holds a different
316 /// number of entries than were sent, and [`Error`] otherwise when the
317 /// batch itself did not reach an answer or its `Bundle` could not be
318 /// decoded. Returns [`Error::UnsupportedParameter`] when one of several
319 /// validations names a parameter the configured release does not
320 /// declare, and then no entry is sent.
321 pub async fn validate_all(
322 &self,
323 validations: &[Validation],
324 ) -> Result<Vec<Result<ValidateOutcome, Error>>, Error> {
325 match validations {
326 [] => Ok(Vec::new()),
327 [one] => Ok(vec![self.validate(one).await]),
328 several => {
329 let requests: Vec<Request> =
330 several.iter().cloned().map(Request::ValidateCode).collect();
331 let version = self.config.wire_version;
332 Ok(self
333 .batch_answers(&requests)
334 .await?
335 .into_iter()
336 .map(|answer| crate::decode::validate_code(version, &answer))
337 .collect())
338 }
339 }
340 }
341
342 /// Sends `requests` as one `batch` `Bundle` and answers them by position.
343 ///
344 /// "The server processes each entry independently" and answers a
345 /// `batch-response` whose entries are "in the same order as the request"
346 /// (<https://hl7.org/fhir/R4/http.html#transaction>), so entry `n` is the
347 /// answer to request `n`. An entry the server refused is that entry's own
348 /// typed error and leaves the others alone.
349 ///
350 /// # Errors
351 /// Returns [`Error::BatchArity`] when the answer holds a different number
352 /// of entries than were sent, and [`Error`] otherwise when the batch
353 /// itself did not reach an answer or its `Bundle` could not be decoded.
354 /// Returns [`Error::UnsupportedParameter`] when a request names a
355 /// parameter the configured release does not declare, and then no entry
356 /// is sent.
357 pub async fn batch(
358 &self,
359 requests: &[Request],
360 ) -> Result<Vec<Result<BatchOutcome, Error>>, Error> {
361 let version = self.config.wire_version;
362 Ok(requests
363 .iter()
364 .zip(self.batch_answers(requests).await?)
365 .map(|(request, answer)| crate::decode::batch_entry(version, request, &answer))
366 .collect())
367 }
368
369 /// Sends `requests` as one `batch` `Bundle` and returns each entry's
370 /// answer, in request order.
371 async fn batch_answers(&self, requests: &[Request]) -> Result<Vec<Answer>, Error> {
372 let version = self.config.wire_version;
373 let body = crate::wire::batch_body(version, requests)
374 .map_err(|source| source.into_error(crate::BATCH, version))?;
375 let answer = self
376 .execute(Call {
377 operation: crate::BATCH,
378 url: self.config.base_url.clone(),
379 body,
380 })
381 .await?;
382 if answer.status != StatusCode::OK {
383 return Err(self.refusal(crate::BATCH, &answer));
384 }
385 let entries =
386 crate::wire::batch_entries(version, &answer.body).map_err(|source| Error::Body {
387 operation: crate::BATCH,
388 url: answer.url.clone(),
389 source: Box::new(source),
390 })?;
391 if entries.len() != requests.len() {
392 return Err(Error::BatchArity {
393 sent: requests.len(),
394 answered: entries.len(),
395 });
396 }
397 Ok(entries
398 .into_iter()
399 .map(|entry| Answer {
400 url: answer.url.clone(),
401 status: entry.status,
402 body: entry.body,
403 })
404 .collect())
405 }
406
407 /// Returns the answer to the `POST` form of `request`.
408 ///
409 /// The operations are invoked with a `Parameters` body, the form that
410 /// carries every input whatever its type
411 /// (<https://hl7.org/fhir/R4/operations.html>).
412 async fn operation(&self, request: &Request) -> Result<Answer, Error> {
413 let operation = request.operation();
414 let version = self.config.wire_version;
415 let body = crate::wire::request_body(version, request)
416 .map_err(|source| source.into_error(operation, version))?;
417 let url = self.url(operation)?;
418 self.execute(Call {
419 operation,
420 url,
421 body,
422 })
423 .await
424 }
425
426 /// Returns the absolute URL of `operation` under the configured base.
427 fn url(&self, operation: &str) -> Result<Url, Error> {
428 let mut url = self.config.base_url.clone();
429 {
430 let mut path = url.path_segments_mut().map_err(|()| Error::BaseUrl {
431 base: self.config.base_url.clone(),
432 })?;
433 path.pop_if_empty();
434 path.extend(operation.split('/'));
435 }
436 Ok(url)
437 }
438
439 /// Returns the answer to `call`, retrying it when the budget allows.
440 async fn execute(&self, call: Call) -> Result<Answer, Error> {
441 let client = self;
442 let request = &call;
443 let attempt = move || async move { client.attempt(request).await };
444 attempt
445 .retry(self.backoff())
446 .when(Error::is_retryable)
447 .notify(|error: &Error, delay| {
448 tracing::warn!(
449 error = error.kind(),
450 delay_ms = delay.as_millis(),
451 "retrying a terminology request"
452 );
453 })
454 .await
455 }
456
457 /// Returns the retry schedule the configuration asks for.
458 fn backoff(&self) -> ExponentialBuilder {
459 let retries =
460 usize::try_from(self.config.retry.max_attempts.saturating_sub(1)).unwrap_or(usize::MAX);
461 ExponentialBuilder::new()
462 .with_min_delay(self.config.retry.initial_backoff)
463 .with_max_delay(self.config.retry.max_backoff)
464 .with_max_times(retries)
465 }
466
467 /// Sends `call` once and reads the whole answer.
468 async fn attempt(&self, call: &Call) -> Result<Answer, Error> {
469 let (response, renewal) = self
470 .send(&call.url, || {
471 self.http
472 .request(Method::POST, call.url.clone())
473 .header(ACCEPT, FHIR_JSON)
474 .header(CONTENT_TYPE, FHIR_JSON)
475 .body(call.body.clone())
476 })
477 .await?;
478
479 let status = response.status();
480 let headers = response.headers().clone();
481 let bytes = response.bytes().await.map_err(|source| Error::Transport {
482 url: call.url.clone(),
483 source,
484 })?;
485 let answer = Answer {
486 url: call.url.clone(),
487 status,
488 body: String::from_utf8_lossy(&bytes).into_owned(),
489 };
490 // A code is a clinical fact, so the log carries the path and never the
491 // body or the query.
492 tracing::debug!(
493 operation = call.operation,
494 path = answer.url.path(),
495 status = %status,
496 "the terminology server answered"
497 );
498
499 if status == StatusCode::UNAUTHORIZED
500 && let Some(source) = renewal
501 {
502 return Err(Error::Renewal {
503 url: answer.url.clone(),
504 upstream: Box::new(self.upstream(&answer)),
505 source,
506 });
507 }
508 if status == StatusCode::UNAUTHORIZED {
509 return Err(Error::Unauthorized {
510 url: answer.url.clone(),
511 challenge: challenge(&headers),
512 upstream: Box::new(self.upstream(&answer)),
513 });
514 }
515 if status.is_server_error() {
516 return Err(Error::ServerFailure {
517 url: answer.url.clone(),
518 upstream: Box::new(self.upstream(&answer)),
519 });
520 }
521 Ok(answer)
522 }
523
524 /// Returns `answer` as an upstream refusal, read in the wire version.
525 fn upstream(&self, answer: &Answer) -> UpstreamError {
526 UpstreamError::new(self.config.wire_version, answer.status, answer.body.clone())
527 }
528
529 /// Returns the error for an answer the operation cannot read.
530 fn refusal(&self, operation: &'static str, answer: &Answer) -> Error {
531 Error::Refused {
532 operation,
533 url: answer.url.clone(),
534 upstream: Box::new(self.upstream(answer)),
535 }
536 }
537}
538
539/// Returns the response to `request`, sent once, for `url`.
540async fn dispatch(url: &Url, request: reqwest::RequestBuilder) -> Result<reqwest::Response, Error> {
541 request.send().await.map_err(|source| {
542 if source.is_timeout() {
543 Error::Timeout {
544 url: url.clone(),
545 source,
546 }
547 } else {
548 Error::Transport {
549 url: url.clone(),
550 source,
551 }
552 }
553 })
554}
555
556/// Returns the `WWW-Authenticate` challenge, when the answer carries usable
557/// text.
558fn challenge(headers: &HeaderMap) -> Option<String> {
559 headers
560 .get(WWW_AUTHENTICATE)
561 .and_then(|value| value.to_str().ok())
562 .map(str::to_owned)
563}
564
565#[cfg(test)]
566mod tests {
567 use super::Client;
568 use crate::config::{Config, WireVersion};
569
570 fn client(base: &str) -> Client {
571 Client::new(Config::new(
572 base.parse().expect("a valid URL"),
573 WireVersion::R4,
574 ))
575 .expect("a buildable client")
576 }
577
578 #[test]
579 fn an_operation_url_sits_under_the_versioned_base() {
580 let url = client("http://tx.invalid/fhir/r4")
581 .url("CodeSystem/$lookup")
582 .expect("a URL under the base");
583 assert_eq!("http://tx.invalid/fhir/r4/CodeSystem/$lookup", url.as_str());
584 }
585
586 #[test]
587 fn a_trailing_slash_on_the_base_does_not_double() {
588 let url = client("http://tx.invalid/r4b/")
589 .url("ValueSet/$validate-code")
590 .expect("a URL under the base");
591 assert_eq!(
592 "http://tx.invalid/r4b/ValueSet/$validate-code",
593 url.as_str()
594 );
595 }
596}