haste_server/
services.rs

1use crate::config::{SearchConfig, SecretProviderConfig, ServerConfig};
2use crate::fhir_client::{FHIRServerClient, ServerClientConfig};
3use haste_fhir_model::r4::generated::terminology::IssueType;
4use haste_fhir_operation_error::{OperationOutcomeError, derive::OperationOutcomeError};
5use haste_fhir_search::elastic_search::search_parameter_resolver::ElasticSearchParameterResolver;
6use haste_fhir_search::{
7    SearchEngine,
8    elastic_search::{ElasticSearchEngine, create_es_client},
9    pg_search::{
10        PgSearchEngine, create_pg_search_pool, search_parameter_resolver::PgSearchParameterResolver,
11    },
12};
13use haste_fhir_terminology::{FHIRTerminology, client::FHIRCanonicalTerminology};
14use haste_fhirpath::FPEngine;
15use haste_repository::{Repository, pg::PGConnection};
16use sqlx::{Pool, Postgres};
17use sqlx_postgres::PgPoolOptions;
18use std::{env::VarError, sync::Arc};
19use tokio::sync::OnceCell;
20use tracing::info;
21
22// Singleton for the database connection pool in postgres.
23static POOL: OnceCell<Pool<Postgres>> = OnceCell::const_new();
24pub async fn get_pool(config: &ServerConfig) -> &'static Pool<Postgres> {
25    match &config.repo {
26        crate::config::RepoConfig::Postgres(pg_config) => {
27            POOL.get_or_init(async || {
28                info!("Connecting to postgres database");
29
30                PgPoolOptions::new()
31                    .max_connections(pg_config.max_connections)
32                    .connect(&pg_config.database_url)
33                    .await
34                    .expect("Failed to create database connection pool")
35            })
36            .await
37        }
38    }
39}
40
41#[derive(OperationOutcomeError, Debug)]
42pub enum ConfigError {
43    #[error(code = "invalid", diagnostic = "Invalid environment!")]
44    DotEnv(#[from] dotenvy::Error),
45    #[error(code = "invalid", diagnostic = "Invalid session!")]
46    Session(#[from] tower_sessions::session::Error),
47    #[error(code = "invalid", diagnostic = "Database error")]
48    Database(#[from] sqlx::Error),
49    #[error(code = "invalid", diagnostic = "Environment variable not set {arg0}")]
50    EnvironmentVariable(#[from] VarError),
51    #[error(code = "invalid", diagnostic = "Failed to render template.")]
52    TemplateRender,
53}
54
55#[derive(OperationOutcomeError, Debug)]
56pub enum CustomOpError {
57    #[error(code = "invalid", diagnostic = "FHIRPath error")]
58    FHIRPath(#[from] haste_fhirpath::FHIRPathError),
59    #[error(code = "invalid", diagnostic = "Failed to deserialize resource")]
60    Deserialize(#[from] serde_json::Error),
61    #[error(code = "invalid", diagnostic = "Internal server error")]
62    InternalServerError,
63}
64
65pub struct ServerState<
66    Repo: Repository + Send + Sync + 'static,
67    Search: SearchEngine + Send + Sync + 'static,
68    Terminology: FHIRTerminology + Send + Sync + 'static,
69> {
70    pub terminology: Arc<Terminology>,
71    pub search: Arc<Search>,
72    pub repo: Arc<Repo>,
73    pub rate_limit: Arc<dyn haste_rate_limit::RateLimit>,
74    pub fhir_client: Arc<FHIRServerClient<Repo, Search, Terminology>>,
75    pub secret_provider: Arc<dyn haste_encryption::SecretsProvider + Send + Sync>,
76    pub config: Arc<crate::config::ServerConfig>,
77}
78
79impl<
80    Repo: Repository + Send + Sync + 'static,
81    Search: SearchEngine + Send + Sync + 'static,
82    Terminology: FHIRTerminology + Send + Sync + 'static,
83> ServerState<Repo, Search, Terminology>
84{
85    pub async fn transaction(&self) -> Result<Self, OperationOutcomeError> {
86        self.repo.transaction(true).await.map(|tx_repo| {
87            let tx_repo = Arc::new(tx_repo);
88            ServerState {
89                terminology: self.terminology.clone(),
90                search: self.search.clone(),
91                repo: tx_repo.clone(),
92                rate_limit: self.rate_limit.clone(),
93                secret_provider: self.secret_provider.clone(),
94                fhir_client: Arc::new(FHIRServerClient::new(ServerClientConfig::new(
95                    tx_repo,
96                    self.search.clone(),
97                    self.terminology.clone(),
98                    self.config.clone(),
99                    self.fhir_client.deno_pool(),
100                ))),
101                config: self.config.clone(),
102            }
103        })
104    }
105
106    pub async fn commit(self) -> Result<(), OperationOutcomeError> {
107        let repo = self.repo.clone();
108        drop(self);
109
110        Arc::try_unwrap(repo)
111            .map_err(|_e| {
112                OperationOutcomeError::fatal(
113                    IssueType::exception(),
114                    "Failed to unwrap transaction client".to_string(),
115                )
116            })?
117            .commit()
118            .await?;
119
120        Ok(())
121    }
122}
123
124/// Enum dispatch for search backends. Allows the server to use either
125/// Elasticsearch or PostgreSQL as the search backend without changing
126/// generic type parameters throughout the codebase.
127#[derive(Clone)]
128pub enum SearchEngineBackend {
129    Elasticsearch(ElasticSearchEngine<ElasticSearchParameterResolver<PGConnection>>),
130    Postgres(PgSearchEngine<PgSearchParameterResolver<PGConnection>>),
131}
132
133impl SearchEngine for SearchEngineBackend {
134    async fn search(
135        &self,
136        fhir_version: &haste_repository::types::SupportedFHIRVersions,
137        tenant: &haste_jwt::TenantId,
138        project: &haste_jwt::ProjectId,
139        search_request: &haste_fhir_client::request::SearchRequest,
140        options: Option<haste_fhir_search::SearchOptions>,
141    ) -> Result<haste_fhir_search::SearchReturn, OperationOutcomeError> {
142        match self {
143            SearchEngineBackend::Elasticsearch(e) => {
144                e.search(fhir_version, tenant, project, search_request, options)
145                    .await
146            }
147            SearchEngineBackend::Postgres(e) => {
148                e.search(fhir_version, tenant, project, search_request, options)
149                    .await
150            }
151        }
152    }
153
154    async fn index(
155        &self,
156        fhir_version: haste_repository::types::SupportedFHIRVersions,
157        resource: Vec<haste_fhir_search::IndexResource>,
158    ) -> Result<haste_fhir_search::IndexOutcome, OperationOutcomeError> {
159        match self {
160            SearchEngineBackend::Elasticsearch(e) => e.index(fhir_version, resource).await,
161            SearchEngineBackend::Postgres(e) => e.index(fhir_version, resource).await,
162        }
163    }
164
165    async fn migrate(
166        &self,
167        fhir_version: &haste_repository::types::SupportedFHIRVersions,
168    ) -> Result<(), OperationOutcomeError> {
169        match self {
170            SearchEngineBackend::Elasticsearch(e) => e.migrate(fhir_version).await,
171            SearchEngineBackend::Postgres(e) => e.migrate(fhir_version).await,
172        }
173    }
174}
175
176async fn create_search_engine(
177    config: &crate::config::ServerConfig,
178    repo: Arc<PGConnection>,
179) -> Result<Arc<SearchEngineBackend>, OperationOutcomeError> {
180    match &config.search {
181        SearchConfig::Elasticsearch(elasticsearch_config) => {
182            let es_client = create_es_client(
183                &elasticsearch_config.url,
184                elasticsearch_config.username.clone(),
185                elasticsearch_config.password.clone(),
186            )?;
187            let engine = ElasticSearchEngine::new(
188                Arc::new(ElasticSearchParameterResolver::new(es_client.clone(), repo)),
189                Arc::new(FPEngine::new()),
190                es_client,
191                elasticsearch_config.prune_removed_search_parameters,
192            );
193            Ok(Arc::new(SearchEngineBackend::Elasticsearch(engine)))
194        }
195        SearchConfig::Postgres(pg_config) => {
196            let search_pool =
197                create_pg_search_pool(&pg_config.database_url, pg_config.max_connections).await?;
198
199            let resolver = PgSearchParameterResolver::new(search_pool.clone(), repo);
200
201            let engine =
202                PgSearchEngine::new(Arc::new(resolver), Arc::new(FPEngine::new()), search_pool);
203            Ok(Arc::new(SearchEngineBackend::Postgres(engine)))
204        }
205    }
206}
207
208pub async fn create_services(
209    config: Arc<crate::config::ServerConfig>,
210) -> Result<
211    Arc<ServerState<PGConnection, SearchEngineBackend, FHIRCanonicalTerminology>>,
212    OperationOutcomeError,
213> {
214    let pool = Arc::new(PGConnection::pool(get_pool(config.as_ref()).await.clone()));
215
216    let terminology = Arc::new(FHIRCanonicalTerminology::new());
217
218    let search_engine = create_search_engine(config.as_ref(), pool.clone()).await?;
219
220    // [TODO] refactor later into generic operation executor.
221    // Cannot just rebuild in each request because it would create multiple DenoPools,
222    // which is expensive and can exhaust system resources.
223    let deno_pool = Arc::new(
224        haste_operation_executor::providers::deno_embedded::pool::DenoPool::new(
225            config.operations.deno_pool_threads,
226        )
227        .expect("Failed to create DenoPool"),
228    );
229
230    let fhir_client = Arc::new(FHIRServerClient::new(
231        ServerClientConfig::new(
232            pool.clone(),
233            search_engine.clone(),
234            terminology.clone(),
235            config.clone(),
236            deno_pool,
237        )
238        .with_mutate_artifacts(config.allow_artifact_mutations)
239        .with_audit_repo(if config.monitoring.audit_enabled {
240            Some(pool.clone())
241        } else {
242            None
243        }),
244    ));
245
246    let shared_state = Arc::new(ServerState {
247        config: config.clone(),
248        rate_limit: pool.clone(),
249        repo: pool,
250        terminology,
251        search: search_engine,
252        fhir_client,
253        secret_provider: match &config.security.encryption {
254            SecretProviderConfig::Environment { prefix } => haste_encryption::get_secrets_provider(
255                haste_encryption::SecretsProviderKind::Environment {
256                    prefix: prefix.clone(),
257                },
258            ),
259            _ => {
260                return Err(OperationOutcomeError::fatal(
261                    IssueType::exception(),
262                    "Only environment encryption is supported for now.".to_string(),
263                ));
264            }
265        },
266    });
267
268    Ok(shared_state)
269}
haste_server/services.rs

haste_server/
services.rs
