haste_fhir_search/pg_search/
mod.rs

1use std::sync::{Arc, LazyLock};
2
3use haste_fhir_client::request::SearchRequest;
4use haste_fhir_model::r4::generated::{resources::Resource, terminology::IssueType};
5use haste_fhir_operation_error::{OperationOutcomeError, derive::OperationOutcomeError};
6use haste_fhirpath::FPEngine;
7use haste_jwt::{ProjectId, TenantId};
8use haste_repository::types::{FHIRMethod, SupportedFHIRVersions};
9use sqlx::{Pool, Postgres, postgres::PgPoolOptions};
10
11use crate::search_parameter_cardinality;
12use crate::{
13    IndexOutcome, IndexResource, ParameterLevel, ResolvedParameter, SearchEngine, SearchOptions,
14    SearchParameterResolve, SearchReturn,
15    elastic_search::is_mapped_search_parameter_type,
16    indexing_conversion::{self, InsertableIndex},
17    memory::R4_SEARCH_PARAMETERS_INDEX,
18    pg_search::schema::{SchemaRegistry, generate_schemas},
19};
20
21mod indexing;
22pub mod keys;
23pub mod migration;
24pub mod schema;
25mod search;
26pub mod search_parameter_resolver;
27
28#[derive(OperationOutcomeError, Debug)]
29pub enum PgSearchError {
30    #[fatal(
31        code = "exception",
32        diagnostic = "Failed to evaluate fhirpath expression."
33    )]
34    FHIRPathError(#[from] haste_fhirpath::FHIRPathError),
35    #[fatal(
36        code = "exception",
37        diagnostic = "PG search does not support the fhir method: '{arg0:?}'"
38    )]
39    UnsupportedFHIRMethod(FHIRMethod),
40    #[fatal(code = "exception", diagnostic = "PG search database error.")]
41    SqlxError(String),
42}
43
44impl From<sqlx::Error> for PgSearchError {
45    fn from(e: sqlx::Error) -> Self {
46        tracing::error!(error = %e, "PG search database error");
47        PgSearchError::SqlxError(e.to_string())
48    }
49}
50
51/// [`SearchEngine`] backed by PostgreSQL. A thin handle: the work is done by
52/// the free functions in [`search`], [`indexing`] and [`migration`].
53#[derive(Clone)]
54pub struct PgSearchEngine<SearchParameterResolver: SearchParameterResolve + 'static> {
55    parameter_resolver: Arc<SearchParameterResolver>,
56    fp_engine: Arc<FPEngine>,
57    pool: Pool<Postgres>,
58    /// Built from the same R4 parameters the migration uses, so the two agree.
59    schema_registry: Arc<SchemaRegistry>,
60}
61
62/// Built once per process; it walks every HL7 `SearchParameter`.
63static R4_SCHEMA_REGISTRY: LazyLock<Arc<SchemaRegistry>> = LazyLock::new(|| {
64    Arc::new(generate_schemas(
65        SupportedFHIRVersions::R4,
66        &R4_SEARCH_PARAMETERS_INDEX.all_parameters(),
67    ))
68});
69
70impl<Resolver: SearchParameterResolve + 'static> PgSearchEngine<Resolver> {
71    pub fn new(
72        parameter_resolver: Arc<Resolver>,
73        fp_engine: Arc<FPEngine>,
74        pool: Pool<Postgres>,
75    ) -> Self {
76        PgSearchEngine {
77            parameter_resolver,
78            fp_engine,
79            pool,
80            schema_registry: R4_SCHEMA_REGISTRY.clone(),
81        }
82    }
83
84    /// Checks the search database responds, for startup waits.
85    ///
86    /// # Errors
87    ///
88    /// Returns an error if the pool cannot run a statement.
89    pub async fn is_connected(&self) -> Result<(), OperationOutcomeError> {
90        sqlx::query("SELECT 1")
91            .execute(&self.pool)
92            .await
93            .map(|_| ())
94            .map_err(|e| {
95                tracing::error!(error = %e, "PG search database is not reachable");
96                OperationOutcomeError::fatal(
97                    IssueType::exception(),
98                    "Search database is not reachable.".to_string(),
99                )
100            })
101    }
102}
103
104/// Creates the search index's own connection pool.
105///
106/// # Errors
107///
108/// Returns an error if the first connection cannot be opened.
109pub async fn create_pg_search_pool(
110    database_url: &str,
111    max_connections: u32,
112) -> Result<Pool<Postgres>, OperationOutcomeError> {
113    PgPoolOptions::new()
114        .max_connections(max_connections)
115        .connect(database_url)
116        .await
117        .map_err(|e| {
118            OperationOutcomeError::fatal(
119                IssueType::exception(),
120                format!("Failed to create PG search database pool: {e}"),
121            )
122        })
123}
124
125/// One resource's evaluated search values, split by storage.
126#[derive(Default)]
127pub(crate) struct ResourceSearchIndex {
128    /// Singular system parameters, keyed by `code`: type table or anchor
129    /// columns.
130    pub system_entries: Vec<(String, InsertableIndex)>,
131    /// Possibly repeating parameters, keyed by URL: shared-table rows.
132    pub dynamic_entries: Vec<(String, InsertableIndex)>,
133}
134
135/// Evaluates each parameter's `FHIRPath` expression against `resource` and
136/// splits the results by where they are stored.
137pub(crate) async fn resource_to_search_index(
138    fp_engine: Arc<FPEngine>,
139    parameters: &[ResolvedParameter],
140    resource: &Resource,
141    resource_type: &str,
142) -> Result<ResourceSearchIndex, OperationOutcomeError> {
143    let mut index = ResourceSearchIndex::default();
144
145    for (param, expression, url) in parameters.iter().filter_map(indexable) {
146        let evaluated = fp_engine
147            .evaluate(expression, vec![resource])
148            .await
149            .map_err(|e| {
150                tracing::error!(
151                    "Failed to evaluate FHIRPath expression: '{}' for resource.",
152                    expression,
153                );
154                PgSearchError::from(e)
155            })?;
156
157        let insertable =
158            indexing_conversion::to_insertable_index(param, &evaluated.iter().collect::<Vec<_>>())?;
159
160        // Cardinality, not level, decides the destination: a column only
161        // exists where a second value is impossible.
162        match param.level {
163            ParameterLevel::System
164                if search_parameter_cardinality::is_single_valued(url, resource_type) =>
165            {
166                if let Some(code) = param.search_parameter.code.value.as_ref() {
167                    index.system_entries.push((code.clone(), insertable));
168                }
169            }
170            ParameterLevel::System | ParameterLevel::Project => {
171                index.dynamic_entries.push((url.clone(), insertable));
172            }
173        }
174    }
175
176    Ok(index)
177}
178
179/// `(parameter, expression, url)` for a parameter that can be stored: it has
180/// both, and a mapped type (not composite or special).
181fn indexable(param: &ResolvedParameter) -> Option<(&ResolvedParameter, &String, &String)> {
182    let search_parameter = &param.search_parameter;
183    let expression = search_parameter.expression.as_ref()?.value.as_ref()?;
184    let url = search_parameter.url.value.as_ref()?;
185
186    is_mapped_search_parameter_type(&search_parameter.type_).then_some((param, expression, url))
187}
188
189impl<SearchParameterResolver: SearchParameterResolve> SearchEngine
190    for PgSearchEngine<SearchParameterResolver>
191{
192    async fn search(
193        &self,
194        _fhir_version: &SupportedFHIRVersions,
195        tenant: &TenantId,
196        project: &ProjectId,
197        search_request: &SearchRequest,
198        options: Option<SearchOptions>,
199    ) -> Result<SearchReturn, OperationOutcomeError> {
200        search::execute_search(
201            &self.pool,
202            self.parameter_resolver.clone(),
203            &self.schema_registry,
204            tenant,
205            project,
206            search_request,
207            options.as_ref(),
208        )
209        .await
210    }
211
212    async fn index(
213        &self,
214        _fhir_version: SupportedFHIRVersions,
215        resources: Vec<IndexResource>,
216    ) -> Result<IndexOutcome, OperationOutcomeError> {
217        // Each part commits before it returns, so it is searchable at once.
218        crate::project_parameters::index_search_parameters_first(
219            self.parameter_resolver.as_ref(),
220            resources,
221            |resources, _searchable| {
222                indexing::index_resources(
223                    &self.pool,
224                    &self.parameter_resolver,
225                    &self.schema_registry,
226                    self.fp_engine.clone(),
227                    resources,
228                )
229            },
230        )
231        .await
232    }
233
234    async fn migrate(
235        &self,
236        _fhir_version: &SupportedFHIRVersions,
237    ) -> Result<(), OperationOutcomeError> {
238        migration::run_migration(&self.pool, &self.schema_registry).await
239    }
240}
haste_fhir_search/pg_search/mod.rs

haste_fhir_search/pg_search/
mod.rs
