haste_fhir_search/
config.rs

1//! Search backend configuration and construction.
2//!
3//! Lives here rather than in each binary so the server and the indexing worker
4//! describe and build the same engine from the same settings. When they each
5//! owned a copy, the worker's only knew about Elasticsearch — so pointing the
6//! server at PostgreSQL left the worker indexing somewhere the server never
7//! read, with nothing to signal it.
8
9use std::sync::Arc;
10
11use haste_fhir_model::r4::generated::terminology::IssueType;
12use haste_fhir_operation_error::OperationOutcomeError;
13use haste_fhirpath::FPEngine;
14use haste_repository::pg::PGConnection;
15use haste_repository::types::SearchIndexBackend;
16use serde::{Deserialize, Serialize};
17
18use crate::{
19    IndexOutcome, IndexResource, SearchEngine, SearchOptions, SearchReturn,
20    elastic_search::{
21        ElasticSearchEngine, create_es_client,
22        search_parameter_resolver::ElasticSearchParameterResolver,
23    },
24    pg_search::{
25        PgSearchEngine, create_pg_search_pool, search_parameter_resolver::PgSearchParameterResolver,
26    },
27};
28
29/// Where the FHIR server stores its search index.
30#[derive(Debug, Clone, Deserialize, Serialize)]
31#[serde(tag = "backend", rename_all = "snake_case")]
32pub enum SearchConfig {
33    Elasticsearch(ElasticsearchConfig),
34    Postgres(PostgresSearchConfig),
35}
36
37#[derive(Debug, Clone, Deserialize, Serialize)]
38pub struct ElasticsearchConfig {
39    pub url: String,
40    pub username: String,
41    pub password: String,
42    /// Lets `migrate search` rebuild the index when a search parameter is
43    /// removed, dropping its mapping and already-indexed data. Elasticsearch
44    /// mappings are append-only, so dropping one needs a reindex. Off by
45    /// default: removed parameters are logged, never deleted.
46    #[serde(default)]
47    pub prune_removed_search_parameters: bool,
48}
49
50#[derive(Debug, Clone, Deserialize, Serialize)]
51pub struct PostgresSearchConfig {
52    pub database_url: String,
53    pub max_connections: u32,
54}
55
56impl Default for ElasticsearchConfig {
57    fn default() -> Self {
58        Self {
59            url: "http://localhost:9200".to_string(),
60            username: "elastic".to_string(),
61            password: "elastic".to_string(),
62            prune_removed_search_parameters: false,
63        }
64    }
65}
66
67impl Default for PostgresSearchConfig {
68    fn default() -> Self {
69        Self {
70            database_url: "postgresql://postgres:postgres@127.0.0.1/haste_health_search"
71                .to_string(),
72            max_connections: 20,
73        }
74    }
75}
76
77impl Default for SearchConfig {
78    fn default() -> Self {
79        SearchConfig::Postgres(PostgresSearchConfig::default())
80    }
81}
82
83/// A configured search engine.
84///
85/// The concrete engines differ in their resolver type, so a caller holding one
86/// generically would have to thread that type everywhere. This dispatches
87/// instead, which is what lets the server and the worker switch backends by
88/// configuration alone.
89#[derive(Clone)]
90pub enum SearchEngineBackend {
91    Elasticsearch(ElasticSearchEngine<ElasticSearchParameterResolver<PGConnection>>),
92    Postgres(PgSearchEngine<PgSearchParameterResolver<PGConnection>>),
93}
94
95impl SearchEngineBackend {
96    /// Whether the backend is reachable, for a caller that waits on it at
97    /// startup.
98    ///
99    /// # Errors
100    ///
101    /// Returns an error if the backend cannot be reached.
102    pub async fn is_connected(&self) -> Result<(), OperationOutcomeError> {
103        match self {
104            SearchEngineBackend::Elasticsearch(engine) => {
105                engine.is_connected().await.map_err(|e| {
106                    OperationOutcomeError::fatal(
107                        IssueType::exception(),
108                        format!("Elasticsearch is not reachable: {e:?}"),
109                    )
110                })
111            }
112            SearchEngineBackend::Postgres(engine) => engine.is_connected().await,
113        }
114    }
115
116    /// What this backend is, for logging that has to say which one it waited
117    /// on.
118    #[must_use]
119    pub const fn name(&self) -> &'static str {
120        match self {
121            SearchEngineBackend::Elasticsearch(_) => "Elasticsearch",
122            SearchEngineBackend::Postgres(_) => "PostgreSQL search",
123        }
124    }
125
126    /// Which `search_index_locks` row this backend's indexing progress is
127    /// tracked under.
128    #[must_use]
129    pub const fn lock_kind(&self) -> SearchIndexBackend {
130        match self {
131            SearchEngineBackend::Elasticsearch(_) => SearchIndexBackend::Elasticsearch,
132            SearchEngineBackend::Postgres(_) => SearchIndexBackend::Postgres,
133        }
134    }
135}
136
137impl SearchEngine for SearchEngineBackend {
138    async fn search(
139        &self,
140        fhir_version: &haste_repository::types::SupportedFHIRVersions,
141        tenant: &haste_jwt::TenantId,
142        project: &haste_jwt::ProjectId,
143        search_request: &haste_fhir_client::request::SearchRequest,
144        options: Option<SearchOptions>,
145    ) -> Result<SearchReturn, OperationOutcomeError> {
146        match self {
147            SearchEngineBackend::Elasticsearch(e) => {
148                e.search(fhir_version, tenant, project, search_request, options)
149                    .await
150            }
151            SearchEngineBackend::Postgres(e) => {
152                e.search(fhir_version, tenant, project, search_request, options)
153                    .await
154            }
155        }
156    }
157
158    async fn index(
159        &self,
160        fhir_version: haste_repository::types::SupportedFHIRVersions,
161        resource: Vec<IndexResource>,
162    ) -> Result<IndexOutcome, OperationOutcomeError> {
163        match self {
164            SearchEngineBackend::Elasticsearch(e) => e.index(fhir_version, resource).await,
165            SearchEngineBackend::Postgres(e) => e.index(fhir_version, resource).await,
166        }
167    }
168
169    async fn migrate(
170        &self,
171        fhir_version: &haste_repository::types::SupportedFHIRVersions,
172    ) -> Result<(), OperationOutcomeError> {
173        match self {
174            SearchEngineBackend::Elasticsearch(e) => e.migrate(fhir_version).await,
175            SearchEngineBackend::Postgres(e) => e.migrate(fhir_version).await,
176        }
177    }
178}
179
180/// Builds the engine `config` describes, against `repo`.
181///
182/// `migrations` says whether this process is allowed to change the index's
183/// shape. The indexing worker only ever writes documents, so it passes false
184/// and never prunes a mapping out from under the server.
185///
186/// # Errors
187///
188/// Returns an error if the backend cannot be reached or configured.
189pub async fn create_search_engine(
190    config: &SearchConfig,
191    repo: Arc<PGConnection>,
192    migrations: bool,
193) -> Result<Arc<SearchEngineBackend>, OperationOutcomeError> {
194    match config {
195        SearchConfig::Elasticsearch(es_config) => {
196            let client = create_es_client(
197                &es_config.url,
198                es_config.username.clone(),
199                es_config.password.clone(),
200            )?;
201
202            let engine = ElasticSearchEngine::new(
203                Arc::new(ElasticSearchParameterResolver::new(client.clone(), repo)),
204                Arc::new(FPEngine::new()),
205                client,
206                migrations && es_config.prune_removed_search_parameters,
207            );
208
209            Ok(Arc::new(SearchEngineBackend::Elasticsearch(engine)))
210        }
211        SearchConfig::Postgres(pg_config) => {
212            let pool =
213                create_pg_search_pool(&pg_config.database_url, pg_config.max_connections).await?;
214
215            let engine = PgSearchEngine::new(
216                Arc::new(PgSearchParameterResolver::new(pool.clone(), repo)),
217                Arc::new(FPEngine::new()),
218                pool,
219            );
220
221            Ok(Arc::new(SearchEngineBackend::Postgres(engine)))
222        }
223    }
224}
haste_fhir_search/config.rs

haste_fhir_search/
config.rs
