haste_server/
services.rs

1use crate::config::{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::SearchEngine;
6use haste_fhir_terminology::{FHIRTerminology, client::FHIRCanonicalTerminology};
7use haste_repository::config::RepoConfig;
8use haste_repository::{Repository, pg::PGConnection};
9use sqlx::{Pool, Postgres};
10use sqlx_postgres::PgPoolOptions;
11use std::{env::VarError, sync::Arc};
12use tokio::sync::OnceCell;
13use tracing::info;
14
15// Singleton for the database connection pool in postgres.
16static POOL: OnceCell<Pool<Postgres>> = OnceCell::const_new();
17pub async fn get_pool(config: &ServerConfig) -> &'static Pool<Postgres> {
18    match &config.repo {
19        RepoConfig::Postgres(pg_config) => {
20            POOL.get_or_init(async || {
21                info!("Connecting to postgres database");
22
23                PgPoolOptions::new()
24                    .max_connections(pg_config.max_connections)
25                    .connect(&pg_config.database_url)
26                    .await
27                    .expect("Failed to create database connection pool")
28            })
29            .await
30        }
31    }
32}
33
34#[derive(OperationOutcomeError, Debug)]
35pub enum ConfigError {
36    #[error(code = "invalid", diagnostic = "Invalid environment!")]
37    DotEnv(#[from] dotenvy::Error),
38    #[error(code = "invalid", diagnostic = "Invalid session!")]
39    Session(#[from] tower_sessions::session::Error),
40    #[error(code = "invalid", diagnostic = "Database error")]
41    Database(#[from] sqlx::Error),
42    #[error(code = "invalid", diagnostic = "Environment variable not set {arg0}")]
43    EnvironmentVariable(#[from] VarError),
44    #[error(code = "invalid", diagnostic = "Failed to render template.")]
45    TemplateRender,
46}
47
48#[derive(OperationOutcomeError, Debug)]
49pub enum CustomOpError {
50    #[error(code = "invalid", diagnostic = "FHIRPath error")]
51    FHIRPath(#[from] haste_fhirpath::FHIRPathError),
52    #[error(code = "invalid", diagnostic = "Failed to deserialize resource")]
53    Deserialize(#[from] serde_json::Error),
54    #[error(code = "invalid", diagnostic = "Internal server error")]
55    InternalServerError,
56}
57
58#[derive(OperationOutcomeError, Debug)]
59pub enum TransactionError {
60    /// `Arc::try_unwrap` on the transaction's repository found another holder.
61    #[fatal(code = "exception", diagnostic = "Transaction client is still in use.")]
62    ClientStillShared,
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(|_| TransactionError::ClientStillShared)?
112            .commit()
113            .await?;
114
115        Ok(())
116    }
117}
118
119pub use haste_fhir_search::config::SearchEngineBackend;
120
121/// The services of the HTTP server. Owns the search index's shape: this is the
122/// one place that may migrate it.
123pub async fn create_services(
124    config: Arc<crate::config::ServerConfig>,
125) -> Result<
126    Arc<ServerState<PGConnection, SearchEngineBackend, FHIRCanonicalTerminology>>,
127    OperationOutcomeError,
128> {
129    build_services(config, true).await
130}
131
132/// The same services for a worker process (one that runs asynchronous jobs):
133/// everything a request has, so a job can go through the FHIR client exactly
134/// as the request that queued it would have, but never a change to the index's
135/// shape under the running server.
136pub async fn create_worker_services(
137    config: Arc<crate::config::ServerConfig>,
138) -> Result<
139    Arc<ServerState<PGConnection, SearchEngineBackend, FHIRCanonicalTerminology>>,
140    OperationOutcomeError,
141> {
142    build_services(config, false).await
143}
144
145async fn build_services(
146    config: Arc<crate::config::ServerConfig>,
147    migrate_search: bool,
148) -> Result<
149    Arc<ServerState<PGConnection, SearchEngineBackend, FHIRCanonicalTerminology>>,
150    OperationOutcomeError,
151> {
152    let pool = Arc::new(PGConnection::pool(get_pool(config.as_ref()).await.clone()));
153
154    let terminology = Arc::new(FHIRCanonicalTerminology::new());
155
156    let search_engine = haste_fhir_search::config::create_search_engine(
157        &config.search,
158        pool.clone(),
159        migrate_search,
160    )
161    .await?;
162
163    // [TODO] refactor later into generic operation executor.
164    // Cannot just rebuild in each request because it would create multiple DenoPools,
165    // which is expensive and can exhaust system resources.
166    let deno_pool = Arc::new(
167        haste_operation_executor::providers::deno_embedded::pool::DenoPool::new(
168            config.operations.deno_pool_threads,
169        )
170        .expect("Failed to create DenoPool"),
171    );
172
173    let fhir_client = Arc::new(FHIRServerClient::new(
174        ServerClientConfig::new(
175            pool.clone(),
176            search_engine.clone(),
177            terminology.clone(),
178            config.clone(),
179            deno_pool,
180        )
181        .with_mutate_artifacts(config.allow_artifact_mutations)
182        .with_audit_repo(if config.monitoring.audit_enabled {
183            Some(pool.clone())
184        } else {
185            None
186        }),
187    ));
188
189    let shared_state = Arc::new(ServerState {
190        config: config.clone(),
191        rate_limit: pool.clone(),
192        repo: pool,
193        terminology,
194        search: search_engine,
195        fhir_client,
196        secret_provider: match &config.security.encryption {
197            SecretProviderConfig::Environment { prefix } => haste_encryption::get_secrets_provider(
198                haste_encryption::SecretsProviderKind::Environment {
199                    prefix: prefix.clone(),
200                },
201            ),
202            _ => {
203                return Err(OperationOutcomeError::fatal(
204                    IssueType::exception(),
205                    "Only environment encryption is supported for now.".to_string(),
206                ));
207            }
208        },
209    });
210
211    Ok(shared_state)
212}
haste_server/services.rs

haste_server/
services.rs
