haste_fhir_search/
project_parameters.rs

1//! A project's own `SearchParameter`s, resolved on top of the HL7 base
2//! parameters.
3//!
4//! A process caches a project's parameters for [`PROJECT_PARAMETERS_TTL`].
5//! Two things refresh a copy sooner: the indexing worker drops a project's
6//! copy once it has indexed one of its `SearchParameter`s
7//! ([`index_search_parameters_first`]), and a lookup by name that misses
8//! rebuilds the copy once, so a new parameter is searchable as soon as it is
9//! indexed. The TTL bounds what neither catches: an update or delete seen by
10//! a server, a list by type (`_include=*`), and a second worker process.
11
12use std::{collections::HashSet, sync::Arc, time::Duration};
13
14use haste_fhir_client::request::{FHIRSearchTypeRequest, SearchRequest};
15use haste_fhir_model::r4::generated::{
16    resources::{Resource, ResourceType},
17    terminology::IssueType,
18};
19use haste_fhir_operation_error::OperationOutcomeError;
20use haste_jwt::{ProjectId, TenantId};
21use haste_repository::{Repository, fhir::CachePolicy};
22use moka::future::{Cache, CacheBuilder};
23
24use crate::{
25    IndexOutcome, IndexResource, MAX_COUNT, ParameterLevel, ResolvedParameter, SearchOptions,
26    SearchParameterResolve, SearchReturn,
27    memory::{R4_SEARCH_PARAMETERS_INDEX, SearchParametersIndex, create_index_map},
28};
29
30/// How long a process trusts its cached copy of a project's parameters.
31///
32/// A refresh is one small search per project per TTL, paid by the first
33/// lookup after expiry, and only for projects being indexed or searched.
34/// A new parameter does not wait for it (see the module doc), so the TTL
35/// only bounds what the refreshes miss: an update or delete seen by a
36/// server, a list by type, and a second worker process. Fifteen minutes
37/// keeps rebuilds rare on the indexing path.
38pub const PROJECT_PARAMETERS_TTL: Duration = Duration::from_mins(15);
39
40/// A search backend's index, as far as the resolver needs it: a type-level
41/// search whose parameters are resolved with the base parameters alone.
42pub trait IndexSearch: Send + Sync {
43    fn search(
44        &self,
45        tenant: &TenantId,
46        project: &ProjectId,
47        request: &SearchRequest,
48        options: &SearchOptions,
49    ) -> impl Future<Output = Result<SearchReturn, OperationOutcomeError>> + Send;
50}
51
52/// Resolves the HL7 base parameters plus each project's own active
53/// `SearchParameter`s, which `Index` finds and `Repo` reads. A base parameter
54/// takes precedence over a project's of the same name.
55#[derive(Clone)]
56pub struct ProjectParameterResolver<Index, Repo> {
57    index: Index,
58    repo: Arc<Repo>,
59    /// Per-project copies, see [`PROJECT_PARAMETERS_TTL`].
60    cache: Cache<(TenantId, ProjectId), Arc<SearchParametersIndex>>,
61}
62
63impl<Index: IndexSearch, Repo: Repository + Send + Sync> ProjectParameterResolver<Index, Repo> {
64    pub fn new(index: Index, repo: Arc<Repo>) -> Self {
65        ProjectParameterResolver {
66            index,
67            repo,
68            cache: CacheBuilder::new(50_000)
69                .time_to_live(PROJECT_PARAMETERS_TTL)
70                .build(),
71        }
72    }
73
74    /// Builds the project's parameter index from its active `SearchParameter`s.
75    ///
76    /// An ordinary `SearchParameter?status=active` search, so `status` is read
77    /// from wherever the backend's schema stores it, as any other search would.
78    async fn build(
79        &self,
80        tenant: &TenantId,
81        project: &ProjectId,
82    ) -> Result<Arc<SearchParametersIndex>, OperationOutcomeError> {
83        let request = SearchRequest::Type(FHIRSearchTypeRequest {
84            resource_type: ResourceType::SearchParameter,
85            parameters: vec![("status".to_string(), vec!["active".to_string()])].into(),
86        });
87        let options = SearchOptions {
88            count_limit: Some(MAX_COUNT),
89            any_of: None,
90        };
91        // Boxed because this is a whole search running inside a parameter
92        // lookup, which itself runs inside a search. Inline, the nested
93        // future types go deep enough that release builds overflow rustc's
94        // query depth limit. Only a rebuild pays the allocation.
95        let found = Box::pin(self.index.search(tenant, project, &request, &options)).await?;
96        let version_ids = found
97            .entries
98            .iter()
99            .map(|entry| &entry.version_id)
100            .collect::<Vec<_>>();
101
102        let parameters = self
103            .repo
104            .read_by_version_ids(tenant, project, &version_ids, CachePolicy::Cache)
105            .await?
106            .into_iter()
107            .filter_map(|resource| match resource {
108                Resource::SearchParameter(parameter) => Some(parameter),
109                _ => None,
110            })
111            .collect();
112
113        Ok(Arc::new(create_index_map(
114            &ParameterLevel::Project,
115            parameters,
116        )))
117    }
118
119    /// The project's cached parameters, or `None` for the system project,
120    /// which has none of its own.
121    async fn cached(
122        &self,
123        tenant: &TenantId,
124        project: &ProjectId,
125    ) -> Result<Option<Arc<SearchParametersIndex>>, OperationOutcomeError> {
126        if let (TenantId::System, ProjectId::System) = (tenant, project) {
127            return Ok(None);
128        }
129
130        // Concurrent misses share one build.
131        self.cache
132            .try_get_with(
133                (tenant.clone(), project.clone()),
134                self.build(tenant, project),
135            )
136            .await
137            .map(Some)
138            .map_err(|e| OperationOutcomeError::fatal(IssueType::exception(), e.to_string()))
139    }
140}
141
142impl<Index: IndexSearch, Repo: Repository + Send + Sync> SearchParameterResolve
143    for ProjectParameterResolver<Index, Repo>
144{
145    async fn by_resource_type(
146        &self,
147        tenant: &TenantId,
148        project: &ProjectId,
149        resource_type: &ResourceType,
150    ) -> Result<Vec<ResolvedParameter>, OperationOutcomeError> {
151        let mut parameters = R4_SEARCH_PARAMETERS_INDEX
152            .by_resource_type(tenant, project, resource_type)
153            .await?;
154        if let Some(index) = self.cached(tenant, project).await? {
155            parameters.extend(
156                index
157                    .by_resource_type(tenant, project, resource_type)
158                    .await?,
159            );
160        }
161        Ok(parameters)
162    }
163
164    async fn by_name(
165        &self,
166        tenant: &TenantId,
167        project: &ProjectId,
168        resource_type: Option<&ResourceType>,
169        code: &str,
170    ) -> Result<Option<ResolvedParameter>, OperationOutcomeError> {
171        if let Some(parameter) = R4_SEARCH_PARAMETERS_INDEX
172            .by_name(tenant, project, resource_type, code)
173            .await?
174        {
175            return Ok(Some(parameter));
176        }
177
178        let Some(index) = self.cached(tenant, project).await? else {
179            return Ok(None);
180        };
181        if let Some(parameter) = index.get(resource_type, code) {
182            return Ok(Some(parameter.clone()));
183        }
184
185        // The name may belong to a `SearchParameter` indexed since the copy
186        // was built. An unknown name costs one rebuild per lookup.
187        let rebuilt = self.build(tenant, project).await?;
188        self.cache
189            .insert((tenant.clone(), project.clone()), rebuilt.clone())
190            .await;
191        Ok(rebuilt.get(resource_type, code).cloned())
192    }
193
194    async fn all(
195        &self,
196        tenant: &TenantId,
197        project: &ProjectId,
198    ) -> Result<Vec<ResolvedParameter>, OperationOutcomeError> {
199        let mut parameters = R4_SEARCH_PARAMETERS_INDEX.all(tenant, project).await?;
200        if let Some(index) = self.cached(tenant, project).await? {
201            parameters.extend(index.all_parameters());
202        }
203        Ok(parameters)
204    }
205
206    async fn invalidate_project(&self, tenant: &TenantId, project: &ProjectId) {
207        self.cache
208            .invalidate(&(tenant.clone(), project.clone()))
209            .await;
210    }
211}
212
213/// Indexes a batch's `SearchParameter` writes before the rest of it, then
214/// drops the cached parameters of their projects, so every other resource in
215/// the batch is indexed with the project's parameters as they now stand.
216///
217/// `index` indexes one part. It is passed `true` for the `SearchParameter`
218/// part, whose writes must be searchable once it returns, since the next
219/// build reads them.
220pub(crate) async fn index_search_parameters_first<Resolver, Index, Indexed>(
221    resolver: &Resolver,
222    resources: Vec<IndexResource>,
223    index: Index,
224) -> Result<IndexOutcome, OperationOutcomeError>
225where
226    Resolver: SearchParameterResolve,
227    Index: Fn(Vec<IndexResource>, bool) -> Indexed,
228    Indexed: Future<Output = Result<IndexOutcome, OperationOutcomeError>>,
229{
230    let is_parameter =
231        |resource: &IndexResource| resource.resource_type == ResourceType::SearchParameter;
232    // Most batches have none, and are indexed as they are.
233    if !resources.iter().any(is_parameter) {
234        return index(resources, false).await;
235    }
236
237    let (parameters, rest): (Vec<_>, Vec<_>) = resources.into_iter().partition(is_parameter);
238    let projects = parameters
239        .iter()
240        .map(|resource| (resource.tenant.clone(), resource.project.clone()))
241        .collect::<HashSet<_>>();
242    let mut outcome = index(parameters, true).await?;
243    for (tenant, project) in &projects {
244        resolver.invalidate_project(tenant, project).await;
245    }
246
247    if !rest.is_empty() {
248        let rest = index(rest, false).await?;
249        outcome.succeeded += rest.succeeded;
250        outcome.failed.extend(rest.failed);
251    }
252    Ok(outcome)
253}
haste_fhir_search/project_parameters.rs

haste_fhir_search/
project_parameters.rs
