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::SearchConfigError;
6use haste_fhir_search::elastic_search::search_parameter_resolver::ElasticSearchParameterResolver;
7use haste_fhir_search::{
8    SearchEngine,
9    elastic_search::{ElasticSearchEngine, create_es_client},
10};
11use haste_fhir_terminology::{FHIRTerminology, client::FHIRCanonicalTerminology};
12use haste_fhirpath::FPEngine;
13use haste_repository::{Repository, pg::PGConnection};
14use sqlx::{Pool, Postgres};
15use sqlx_postgres::PgPoolOptions;
16use std::{env::VarError, sync::Arc};
17use tokio::sync::OnceCell;
18use tracing::info;
19
20// Singleton for the database connection pool in postgres.
21static POOL: OnceCell<Pool<Postgres>> = OnceCell::const_new();
22pub async fn get_pool(config: &ServerConfig) -> &'static Pool<Postgres> {
23    match &config.repo {
24        crate::config::RepoConfig::Postgres(pg_config) => {
25            POOL.get_or_init(async || {
26                info!("Connecting to postgres database");
27
28                PgPoolOptions::new()
29                    .max_connections(pg_config.max_connections)
30                    .connect(&pg_config.database_url)
31                    .await
32                    .expect("Failed to create database connection pool")
33            })
34            .await
35        }
36    }
37}
38
39#[derive(OperationOutcomeError, Debug)]
40pub enum ConfigError {
41    #[error(code = "invalid", diagnostic = "Invalid environment!")]
42    DotEnv(#[from] dotenvy::Error),
43    #[error(code = "invalid", diagnostic = "Invalid session!")]
44    Session(#[from] tower_sessions::session::Error),
45    #[error(code = "invalid", diagnostic = "Database error")]
46    Database(#[from] sqlx::Error),
47    #[error(code = "invalid", diagnostic = "Environment variable not set {arg0}")]
48    EnvironmentVariable(#[from] VarError),
49    #[error(code = "invalid", diagnostic = "Failed to render template.")]
50    TemplateRender,
51}
52
53#[derive(OperationOutcomeError, Debug)]
54pub enum CustomOpError {
55    #[error(code = "invalid", diagnostic = "FHIRPath error")]
56    FHIRPath(#[from] haste_fhirpath::FHIRPathError),
57    #[error(code = "invalid", diagnostic = "Failed to deserialize resource")]
58    Deserialize(#[from] serde_json::Error),
59    #[error(code = "invalid", diagnostic = "Internal server error")]
60    InternalServerError,
61}
62
63pub struct ServerState<
64    Repo: Repository + Send + Sync + 'static,
65    Search: SearchEngine + Send + Sync + 'static,
66    Terminology: FHIRTerminology + Send + Sync + 'static,
67> {
68    pub terminology: Arc<Terminology>,
69    pub search: Arc<Search>,
70    pub repo: Arc<Repo>,
71    pub rate_limit: Arc<dyn haste_rate_limit::RateLimit>,
72    pub fhir_client: Arc<FHIRServerClient<Repo, Search, Terminology>>,
73    pub secret_provider: Arc<dyn haste_encryption::SecretsProvider + Send + Sync>,
74    pub config: Arc<crate::config::ServerConfig>,
75}
76
77impl<
78    Repo: Repository + Send + Sync + 'static,
79    Search: SearchEngine + Send + Sync + 'static,
80    Terminology: FHIRTerminology + Send + Sync + 'static,
81> ServerState<Repo, Search, Terminology>
82{
83    pub async fn transaction(&self) -> Result<Self, OperationOutcomeError> {
84        self.repo.transaction(true).await.map(|tx_repo| {
85            let tx_repo = Arc::new(tx_repo);
86            ServerState {
87                terminology: self.terminology.clone(),
88                search: self.search.clone(),
89                repo: tx_repo.clone(),
90                rate_limit: self.rate_limit.clone(),
91                secret_provider: self.secret_provider.clone(),
92                fhir_client: Arc::new(FHIRServerClient::new(ServerClientConfig::new(
93                    tx_repo,
94                    self.search.clone(),
95                    self.terminology.clone(),
96                    self.config.clone(),
97                ))),
98                config: self.config.clone(),
99            }
100        })
101    }
102
103    pub async fn commit(self) -> Result<(), OperationOutcomeError> {
104        let repo = self.repo.clone();
105        drop(self);
106
107        Arc::try_unwrap(repo)
108            .map_err(|_e| {
109                OperationOutcomeError::fatal(
110                    IssueType::exception(),
111                    "Failed to unwrap transaction client".to_string(),
112                )
113            })?
114            .commit()
115            .await?;
116
117        Ok(())
118    }
119}
120
121fn create_search_engine<Repo: Repository + Send + Sync + 'static>(
122    config: &crate::config::ServerConfig,
123    parameter_resolver: Arc<Repo>,
124) -> Result<Arc<ElasticSearchEngine<ElasticSearchParameterResolver<Repo>>>, SearchConfigError> {
125    match &config.search {
126        SearchConfig::Elasticsearch(elasticsearch_config) => {
127            let es_client = create_es_client(
128                &elasticsearch_config.url,
129                elasticsearch_config.username.clone(),
130                elasticsearch_config.password.clone(),
131            )?;
132            let k = Arc::new(haste_fhir_search::elastic_search::ElasticSearchEngine::new(
133                Arc::new(ElasticSearchParameterResolver::new(
134                    es_client.clone(),
135                    parameter_resolver,
136                )),
137                Arc::new(FPEngine::new()),
138                es_client,
139            ));
140
141            Ok(k)
142        }
143    }
144}
145
146pub async fn create_services(
147    config: Arc<crate::config::ServerConfig>,
148) -> Result<
149    Arc<
150        ServerState<
151            PGConnection,
152            ElasticSearchEngine<ElasticSearchParameterResolver<PGConnection>>,
153            FHIRCanonicalTerminology,
154        >,
155    >,
156    OperationOutcomeError,
157> {
158    let pool = Arc::new(PGConnection::pool(get_pool(config.as_ref()).await.clone()));
159
160    let terminology = Arc::new(FHIRCanonicalTerminology::new());
161
162    let search_engine = create_search_engine(config.as_ref(), pool.clone())?;
163
164    let fhir_client = Arc::new(FHIRServerClient::new(
165        ServerClientConfig::new(
166            pool.clone(),
167            search_engine.clone(),
168            terminology.clone(),
169            config.clone(),
170        )
171        .with_mutate_artifacts(config.allow_artifact_mutations)
172        .with_audit_repo(if config.monitoring.audit_enabled {
173            Some(pool.clone())
174        } else {
175            None
176        }),
177    ));
178
179    let shared_state = Arc::new(ServerState {
180        config: config.clone(),
181        rate_limit: pool.clone(),
182        repo: pool,
183        terminology,
184        search: search_engine,
185        fhir_client,
186        secret_provider: match &config.security.encryption {
187            SecretProviderConfig::Environment { prefix } => haste_encryption::get_secrets_provider(
188                haste_encryption::SecretsProviderKind::Environment {
189                    prefix: prefix.clone(),
190                },
191            ),
192            _ => {
193                return Err(OperationOutcomeError::fatal(
194                    IssueType::exception(),
195                    "Only environment encryption is supported for now.".to_string(),
196                ));
197            }
198        },
199    });
200
201    Ok(shared_state)
202}
haste_server/services.rs

haste_server/
services.rs
