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::{
12    IndexOutcome, IndexResource, ParameterLevel, ResolvedParameter, SearchEngine, SearchOptions,
13    SearchParameterResolve, SearchReturn,
14    elastic_search::is_mapped_search_parameter_type,
15    indexing_conversion::{self, InsertableIndex},
16    memory::R4_SEARCH_PARAMETERS_INDEX,
17    pg_search::schema::{SchemaRegistry, generate_schemas},
18};
19
20mod indexing;
21pub mod migration;
22pub mod schema;
23mod search;
24pub mod search_parameter_resolver;
25
26#[derive(OperationOutcomeError, Debug)]
27pub enum PgSearchError {
28    #[fatal(
29        code = "exception",
30        diagnostic = "Failed to evaluate fhirpath expression."
31    )]
32    FHIRPathError(#[from] haste_fhirpath::FHIRPathError),
33    #[fatal(
34        code = "exception",
35        diagnostic = "PG search does not support the fhir method: '{arg0:?}'"
36    )]
37    UnsupportedFHIRMethod(FHIRMethod),
38    #[fatal(code = "exception", diagnostic = "PG search database error: '{arg0}'")]
39    SqlxError(String),
40}
41
42impl From<sqlx::Error> for PgSearchError {
43    fn from(e: sqlx::Error) -> Self {
44        PgSearchError::SqlxError(e.to_string())
45    }
46}
47
48#[derive(Clone)]
49pub struct PgSearchEngine<SearchParameterResolver: SearchParameterResolve + 'static> {
50    parameter_resolver: Arc<SearchParameterResolver>,
51    fp_engine: Arc<FPEngine>,
52    pool: Pool<Postgres>,
53    /// Per-resource-type table layouts for the HL7 base search parameters.
54    /// Derived once from the static R4 parameter set, which is the same source
55    /// the migration builds the tables from, so the two can't drift.
56    schema_registry: Arc<SchemaRegistry>,
57}
58
59/// The per-resource-type schemas for the R4 base search parameters. Built once
60/// and shared by every engine instance — deriving them walks every HL7
61/// `SearchParameter`, which is wasted work to repeat.
62static R4_SCHEMA_REGISTRY: LazyLock<Arc<SchemaRegistry>> = LazyLock::new(|| {
63    Arc::new(generate_schemas(
64        &R4_SEARCH_PARAMETERS_INDEX.all_parameters(),
65    ))
66});
67
68/// Creates a separate PostgreSQL connection pool for the search index database.
69///
70/// # Errors
71///
72/// Returns an error if the pool cannot open its first connection — an
73/// unreachable host, bad credentials, or a malformed `database_url`.
74pub async fn create_pg_search_pool(
75    database_url: &str,
76    max_connections: u32,
77) -> Result<Pool<Postgres>, OperationOutcomeError> {
78    PgPoolOptions::new()
79        .max_connections(max_connections)
80        .connect(database_url)
81        .await
82        .map_err(|e| {
83            OperationOutcomeError::fatal(
84                IssueType::exception(),
85                format!("Failed to create PG search database pool: {e}"),
86            )
87        })
88}
89
90impl<SearchParameterResolver: SearchParameterResolve + 'static>
91    PgSearchEngine<SearchParameterResolver>
92{
93    pub fn new(
94        parameter_resolver: Arc<SearchParameterResolver>,
95        fp_engine: Arc<FPEngine>,
96        pool: Pool<Postgres>,
97    ) -> Self {
98        PgSearchEngine {
99            parameter_resolver,
100            fp_engine,
101            pool,
102            schema_registry: R4_SCHEMA_REGISTRY.clone(),
103        }
104    }
105}
106
107/// A single resource's evaluated search values, split the way the hybrid
108/// schema stores them.
109pub(crate) struct ResourceSearchIndex {
110    /// System-level parameters, keyed by search parameter `code`. These land
111    /// in dedicated columns on the per-resource-type table, which is looked up
112    /// by code rather than by URL.
113    pub system_entries: Vec<(String, InsertableIndex)>,
114    /// Project-level parameters, keyed by canonical URL. These land in the
115    /// `search_dynamic_*` EAV tables, where `param_url` discriminates them.
116    pub dynamic_entries: Vec<(String, InsertableIndex)>,
117}
118
119/// Evaluates `FHIRPath` expressions for all applicable search parameters and
120/// converts results into `InsertableIndex` values.
121///
122/// Backend-agnostic beyond the routing: reuses
123/// `indexing_conversion::to_insertable_index` and `FPEngine` from the shared
124/// crate, then splits the results by `ParameterLevel` so each half can be
125/// written to the storage that fits it.
126pub(crate) async fn resource_to_search_index(
127    fp_engine: Arc<FPEngine>,
128    parameters: &[ResolvedParameter],
129    resource: &Resource,
130) -> Result<ResourceSearchIndex, OperationOutcomeError> {
131    let mut system_entries = Vec::new();
132    let mut dynamic_entries = Vec::new();
133
134    for param in parameters {
135        if let Some(expression) = param
136            .search_parameter
137            .expression
138            .as_ref()
139            .and_then(|e| e.value.as_ref())
140            && let Some(url) = param.search_parameter.url.value.as_ref()
141        {
142            // A parameter of an unmapped type (composite, special, ...) has
143            // nowhere to be written, so evaluating it would only discard the
144            // result.
145            if !is_mapped_search_parameter_type(&param.search_parameter.type_) {
146                continue;
147            }
148
149            let result = fp_engine
150                .evaluate(expression, vec![resource])
151                .await
152                .map_err(PgSearchError::from);
153
154            if let Err(err) = result {
155                tracing::error!(
156                    "Failed to evaluate FHIRPath expression: '{}' for resource.",
157                    expression,
158                );
159                return Err(err.into());
160            }
161
162            let insertable = indexing_conversion::to_insertable_index(
163                param,
164                &result?.iter().collect::<Vec<_>>(),
165            )?;
166
167            match &param.level {
168                // Keyed by code: the per-resource-type table's columns are
169                // derived from the code, not the URL.
170                ParameterLevel::System => {
171                    if let Some(code) = param.search_parameter.code.value.as_ref() {
172                        system_entries.push((code.clone(), insertable));
173                    }
174                }
175                ParameterLevel::Project => {
176                    dynamic_entries.push((url.clone(), insertable));
177                }
178            }
179        }
180    }
181
182    Ok(ResourceSearchIndex {
183        system_entries,
184        dynamic_entries,
185    })
186}
187
188impl<SearchParameterResolver: SearchParameterResolve> SearchEngine
189    for PgSearchEngine<SearchParameterResolver>
190{
191    async fn search(
192        &self,
193        _fhir_version: &SupportedFHIRVersions,
194        tenant: &TenantId,
195        project: &ProjectId,
196        search_request: &SearchRequest,
197        options: Option<SearchOptions>,
198    ) -> Result<SearchReturn, OperationOutcomeError> {
199        search::execute_search(
200            &self.pool,
201            self.parameter_resolver.clone(),
202            &self.schema_registry,
203            tenant,
204            project,
205            search_request,
206            options.as_ref(),
207        )
208        .await
209    }
210
211    async fn index(
212        &self,
213        _fhir_version: SupportedFHIRVersions,
214        resources: Vec<IndexResource>,
215    ) -> Result<IndexOutcome, OperationOutcomeError> {
216        indexing::index_resources(
217            &self.pool,
218            &self.parameter_resolver,
219            &self.schema_registry,
220            self.fp_engine.clone(),
221            resources,
222        )
223        .await
224    }
225
226    async fn migrate(
227        &self,
228        _fhir_version: &SupportedFHIRVersions,
229    ) -> Result<(), OperationOutcomeError> {
230        migration::run_migration(&self.pool, &self.schema_registry).await
231    }
232}
haste_fhir_search/pg_search/mod.rs

haste_fhir_search/pg_search/
mod.rs
