haste_fhir_search/memory/search/
mod.rs

1//! A search backend over resources held in memory: the ones the server ships
2//! rather than stores, which no index holds. Parameters get the semantics the
3//! Postgres and Elasticsearch backends give them; what only a query can do
4//! (follow a chain, `_sort`) is refused as not supported.
5//!
6//! A search is a pipeline of pure functions over a snapshot of the resources:
7//! the parameters become criteria and a page, each resource is tested against
8//! the criteria, and the page is cut from the matches. Only evaluating a
9//! parameter's `FHIRPath` expression is async, so the pass over the resources
10//! goes through the small async combinators at the end of this file. They
11//! take an `async fn` and its context rather than a closure: a closure held
12//! across an await can't be proven `Send` (rust-lang/rust#102211).
13
14mod predicate;
15
16use std::sync::{Arc, PoisonError, RwLock};
17
18use crate::{
19    IndexOutcome, IndexResource, MAX_COUNT, ResolvedParameter, SearchEngine, SearchEntry,
20    SearchOptions, SearchReturn,
21    indexing_conversion::to_insertable_index,
22    memory::R4_SEARCH_PARAMETERS_INDEX,
23    query::{QueryBuildError, max_count, parse_u64, type_filter, value::Values},
24};
25use haste_fhir_client::{
26    request::SearchRequest,
27    url::{Parameter, ParsedParameter, ParsedParameters},
28};
29use haste_fhir_model::r4::generated::resources::{Resource, ResourceType};
30use haste_fhir_operation_error::{OperationOutcomeError, derive::OperationOutcomeError};
31use haste_fhirpath::FPEngine;
32use haste_jwt::{ProjectId, ResourceId, TenantId, VersionId};
33use haste_reflect::MetaValue;
34use haste_repository::types::{FHIRMethod, SupportedFHIRVersions};
35use predicate::Predicate;
36
37#[derive(OperationOutcomeError, Debug)]
38enum MemorySearchError {
39    #[fatal(
40        code = "exception",
41        diagnostic = "Failed to evaluate fhirpath expression."
42    )]
43    FHIRPath(#[from] haste_fhirpath::FHIRPathError),
44}
45
46/// A resource and what a search returns for it.
47#[derive(Clone)]
48struct Stored {
49    id: ResourceId,
50    version_id: VersionId,
51    resource_type: ResourceType,
52    resource: Resource,
53}
54
55/// One resource parameter: the expression that finds a resource's values,
56/// and what they must satisfy. The definition is borrowed from the R4
57/// parameter index, which lives for the program; the values from the request.
58struct Criterion<'a> {
59    definition: &'static ResolvedParameter,
60    expression: &'static str,
61    predicate: Predicate<'a>,
62}
63
64/// What a resource must satisfy: every criterion, and one of `any_of`'s when
65/// the search restricts it.
66struct Criteria<'a> {
67    all: Vec<Criterion<'a>>,
68    any_of: Option<Vec<Criterion<'a>>>,
69}
70
71/// The page the result parameters select. The total is always known in
72/// memory, so `_total` only has its value checked.
73struct Page {
74    count: u64,
75    offset: u64,
76}
77
78/// Resources held at the system level: the same in every tenant and project.
79pub struct MemorySearch {
80    resources: RwLock<Arc<Vec<Stored>>>,
81}
82
83impl MemorySearch {
84    /// A backend over `resources`, searched by the R4 search parameters. Each
85    /// is stored as version `1`; one without an id is left out, since no
86    /// search could return it.
87    #[must_use]
88    pub fn new(resources: impl IntoIterator<Item = Resource>) -> Self {
89        let stored = resources
90            .into_iter()
91            .filter_map(|resource| stored(resource, VersionId::new("1".to_string())))
92            .collect();
93
94        MemorySearch {
95            resources: RwLock::new(Arc::new(stored)),
96        }
97    }
98}
99
100impl SearchEngine for MemorySearch {
101    async fn search(
102        &self,
103        _fhir_version: &SupportedFHIRVersions,
104        _tenant: &TenantId,
105        project: &ProjectId,
106        search_request: &SearchRequest,
107        options: Option<SearchOptions>,
108    ) -> Result<SearchReturn, OperationOutcomeError> {
109        // A snapshot, so the lock isn't held while expressions evaluate.
110        let resources = self
111            .resources
112            .read()
113            .unwrap_or_else(PoisonError::into_inner)
114            .clone();
115        search(&resources, project, search_request, options).await
116    }
117
118    /// Adds, replaces or removes each resource by type and id. The resources
119    /// are system-level, so the tenant and project aren't recorded.
120    fn index(
121        &self,
122        _fhir_version: SupportedFHIRVersions,
123        resources: Vec<IndexResource>,
124    ) -> impl Future<Output = Result<IndexOutcome, OperationOutcomeError>> + Send {
125        let succeeded = resources.len();
126        {
127            let mut current = self
128                .resources
129                .write()
130                .unwrap_or_else(PoisonError::into_inner);
131            let next = indexed(&current, resources);
132            *current = Arc::new(next);
133        }
134
135        std::future::ready(Ok(IndexOutcome {
136            succeeded,
137            failed: Vec::new(),
138        }))
139    }
140
141    /// Nothing to migrate: there is no index.
142    fn migrate(
143        &self,
144        _fhir_version: &SupportedFHIRVersions,
145    ) -> impl Future<Output = Result<(), OperationOutcomeError>> + Send {
146        std::future::ready(Ok(()))
147    }
148}
149
150/// `resource` stored at `version_id`; `None` for one without an id, which no
151/// search could return.
152fn stored(resource: Resource, version_id: VersionId) -> Option<Stored> {
153    let id = resource
154        .get_field("id")?
155        .as_any()
156        .downcast_ref::<String>()?
157        .clone();
158    Some(Stored {
159        id: ResourceId::new(id),
160        version_id,
161        resource_type: resource.resource_type(),
162        resource,
163    })
164}
165
166/// `existing` with each of `batch` added, replaced or removed by type and id,
167/// in order.
168fn indexed(existing: &[Stored], batch: Vec<IndexResource>) -> Vec<Stored> {
169    batch
170        .into_iter()
171        .fold(existing.to_vec(), |mut stored, resource| {
172            stored.retain(|current| {
173                current.resource_type != resource.resource_type
174                    || current.id.as_ref() != resource.id.as_ref()
175            });
176            if !matches!(resource.fhir_method, FHIRMethod::Delete) {
177                stored.push(Stored {
178                    id: resource.id,
179                    version_id: resource.version_id,
180                    resource_type: resource.resource_type,
181                    resource: resource.resource,
182                });
183            }
184            stored
185        })
186}
187
188/// The page of `resources` the request selects, with the total. The entries
189/// are stamped with `project`: the resources are the same in every project.
190async fn search(
191    resources: &[Stored],
192    project: &ProjectId,
193    request: &SearchRequest,
194    options: Option<SearchOptions>,
195) -> Result<SearchReturn, OperationOutcomeError> {
196    let (resource_type, parameters) = request_parts(request);
197    let count = max_count(options.as_ref())?;
198    let any_of = options.as_ref().and_then(|options| options.any_of.as_ref());
199
200    let criteria = Criteria {
201        all: parameters
202            .parameters()
203            .iter()
204            .filter_map(resource_parameter)
205            .map(|parameter| criterion(resource_type, parameter))
206            .collect::<Result<_, _>>()?,
207        // `ANDed` with the client's parameters; empty matches nothing.
208        any_of: any_of
209            .map(|any_of| {
210                any_of
211                    .0
212                    .iter()
213                    .map(|parameter| criterion(resource_type, parameter))
214                    .collect::<Result<_, _>>()
215            })
216            .transpose()?,
217    };
218    let page = parameters
219        .parameters()
220        .iter()
221        .filter_map(result_parameter)
222        .try_fold(Page { count, offset: 0 }, paged)?;
223
224    let types = type_filter(request)?;
225    let candidates = resources
226        .iter()
227        .filter(|stored| resource_type.is_none_or(|rt| *rt == stored.resource_type))
228        .filter(|stored| {
229            types
230                .as_ref()
231                .is_none_or(|types| types.contains(&stored.resource_type))
232        })
233        .collect::<Vec<_>>();
234    let hits = retained(candidates, &criteria, matched).await?;
235
236    Ok(SearchReturn {
237        total: Some(i64::try_from(hits.len()).unwrap_or(i64::MAX)),
238        entries: hits
239            .into_iter()
240            .skip(to_usize(page.offset))
241            .take(to_usize(page.count))
242            .map(|stored| entry(stored, project))
243            .collect(),
244    })
245}
246
247fn request_parts(request: &SearchRequest) -> (Option<&ResourceType>, &ParsedParameters) {
248    match request {
249        SearchRequest::Type(request) => (Some(&request.resource_type), &request.parameters),
250        SearchRequest::System(request) => (None, &request.parameters),
251    }
252}
253
254fn resource_parameter(parameter: &ParsedParameter) -> Option<&Parameter> {
255    match parameter {
256        ParsedParameter::Resource(parameter) => Some(parameter),
257        ParsedParameter::Result(_) => None,
258    }
259}
260
261fn result_parameter(parameter: &ParsedParameter) -> Option<&Parameter> {
262    match parameter {
263        ParsedParameter::Result(parameter) => Some(parameter),
264        ParsedParameter::Resource(_) => None,
265    }
266}
267
268/// The criterion for one resource parameter on `resource_type`.
269fn criterion<'a>(
270    resource_type: Option<&ResourceType>,
271    parameter: &'a Parameter,
272) -> Result<Criterion<'a>, OperationOutcomeError> {
273    // Following a reference needs the referenced resource, which only a
274    // query has.
275    if parameter.is_chained() {
276        return Err(QueryBuildError::UnsupportedParameter(parameter.key()).into());
277    }
278
279    let definition = R4_SEARCH_PARAMETERS_INDEX
280        .get(resource_type, &parameter.name)
281        .ok_or_else(|| QueryBuildError::MissingParameter(parameter.name.clone()))?;
282    let search_parameter = &definition.search_parameter;
283    let expression = search_parameter
284        .expression
285        .as_ref()
286        .and_then(|expression| expression.value.as_deref())
287        .ok_or_else(|| QueryBuildError::UnsupportedParameter(parameter.name.clone()))?;
288
289    Ok(Criterion {
290        definition,
291        expression,
292        predicate: Predicate::new(Values::parse(parameter, &search_parameter.type_)?)?,
293    })
294}
295
296/// `page` with one result parameter applied.
297fn paged(page: Page, parameter: &Parameter) -> Result<Page, OperationOutcomeError> {
298    Ok(match parameter.name.as_str() {
299        "_count" => Page {
300            count: parse_u64(parameter)?.min(MAX_COUNT),
301            ..page
302        },
303        "_offset" => Page {
304            offset: parse_u64(parameter)?,
305            ..page
306        },
307        "_total" => match parameter.value.as_slice() {
308            [value] if matches!(value.as_str(), "none" | "estimate" | "accurate") => page,
309            _ => return Err(QueryBuildError::InvalidParameterValue(parameter.name.clone()).into()),
310        },
311        // Handled in middleware, as for the other backends.
312        "_summary" | "_elements" | "_include" | "_type" | "_format" | "_pretty" => page,
313        // `_sort` and the rest need what only a query has.
314        _ => return Err(QueryBuildError::UnsupportedParameter(parameter.name.clone()).into()),
315    })
316}
317
318/// Whether `stored` satisfies the criteria.
319async fn matched(stored: &Stored, criteria: &Criteria<'_>) -> Result<bool, OperationOutcomeError> {
320    let resource = &stored.resource;
321    Ok(every(&criteria.all, resource, matches).await?
322        && if let Some(any_of) = &criteria.any_of {
323            some(any_of, resource, matches).await?
324        } else {
325            true
326        })
327}
328
329/// Whether `resource`'s values for the criterion's parameter satisfy the
330/// searched values.
331async fn matches(
332    criterion: &Criterion<'_>,
333    resource: &Resource,
334) -> Result<bool, OperationOutcomeError> {
335    let found = FPEngine::new()
336        .evaluate(criterion.expression, vec![resource])
337        .await
338        .map_err(MemorySearchError::from)?;
339    let index = to_insertable_index(criterion.definition, &found.iter().collect::<Vec<_>>())?;
340    Ok(criterion.predicate.holds(&index))
341}
342
343/// What a search returns for `stored`, in `project`.
344fn entry(stored: &Stored, project: &ProjectId) -> SearchEntry {
345    SearchEntry {
346        id: stored.id.clone(),
347        resource_type: stored.resource_type.clone(),
348        version_id: stored.version_id.clone(),
349        project: project.clone(),
350    }
351}
352
353/// A page bound as a position in the matches; saturates on a 32-bit target.
354fn to_usize(bound: u64) -> usize {
355    usize::try_from(bound).unwrap_or(usize::MAX)
356}
357
358/// Whether `test` holds for every item, stopping at the first it fails.
359async fn every<T, C: Copy, E>(
360    items: impl IntoIterator<Item = T>,
361    context: C,
362    test: impl AsyncFn(T, C) -> Result<bool, E>,
363) -> Result<bool, E> {
364    for item in items {
365        if !test(item, context).await? {
366            return Ok(false);
367        }
368    }
369    Ok(true)
370}
371
372/// Whether `test` holds for some item, stopping at the first it passes.
373async fn some<T, C: Copy, E>(
374    items: impl IntoIterator<Item = T>,
375    context: C,
376    test: impl AsyncFn(T, C) -> Result<bool, E>,
377) -> Result<bool, E> {
378    for item in items {
379        if test(item, context).await? {
380            return Ok(true);
381        }
382    }
383    Ok(false)
384}
385
386/// The items `keep` holds for, in order.
387async fn retained<T: Copy, C: Copy, E>(
388    items: impl IntoIterator<Item = T>,
389    context: C,
390    keep: impl AsyncFn(T, C) -> Result<bool, E>,
391) -> Result<Vec<T>, E> {
392    let mut accepted = Vec::new();
393    for item in items {
394        if keep(item, context).await? {
395            accepted.push(item);
396        }
397    }
398    Ok(accepted)
399}
haste_fhir_search/memory/search/mod.rs

haste_fhir_search/memory/search/
mod.rs
