haste_server/
load_artifacts.rs

1use std::{collections::HashSet, sync::Arc};
2
3use crate::{config::ServerConfig, fhir_client::ServerCTX, services::create_services};
4use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
5use haste_artifacts::ARTIFACT_RESOURCES;
6use haste_fhir_client::{
7    FHIRClient,
8    request::{FHIRSearchTypeRequest, SearchRequest},
9};
10use haste_fhir_model::r4::generated::{
11    resources::{
12        Bundle, BundleEntry, BundleEntryRequest, OperationDefinition, Resource, ResourceType,
13        SearchParameter, StructureDefinition,
14    },
15    terminology::{BundleType, HttpVerb, IssueType},
16    types::{Coding, FHIRCode, FHIRUri, Meta},
17};
18use haste_fhir_operation_error::OperationOutcomeError;
19use haste_fhir_search::{MAX_COUNT, SearchEngine, SearchOptions};
20use haste_jwt::{ProjectId, TenantId};
21use haste_repository::{Repository, fhir::CachePolicy, types::SupportedFHIRVersions};
22use sha1::{Digest, Sha1};
23
24fn generate_sha256_hash(value: &Resource) -> String {
25    let json = serde_json::to_string(value).expect("failed to serialize value.");
26    let mut sha_hasher = Sha1::new();
27    sha_hasher.update(json.as_bytes());
28    let sha1 = sha_hasher.finalize();
29
30    URL_SAFE_NO_PAD.encode(sha1)
31}
32
33static HASH_TAG_SYSTEM: &str = "https://haste.health/fhir/CodeSystem/hash";
34
35fn _add_hash_tag(meta: &mut Option<Box<Meta>>, sha_hash: String) {
36    let hash_tag = Coding {
37        system: Some(Box::new(FHIRUri {
38            value: Some(HASH_TAG_SYSTEM.to_string()),
39            ..Default::default()
40        })),
41        code: Some(Box::new(FHIRCode {
42            value: Some(sha_hash),
43            ..Default::default()
44        })),
45        ..Default::default()
46    };
47
48    let meta = if let Some(meta) = meta {
49        meta
50    } else {
51        *meta = Some(Box::new(Meta::default()));
52        meta.as_mut().unwrap()
53    };
54
55    match &mut meta.tag {
56        Some(tags) => tags.push(hash_tag),
57        None => meta.tag = Some(vec![hash_tag]),
58    }
59}
60
61fn add_hash_tag(resource: &mut Resource, sha_hash: String) {
62    match resource {
63        Resource::StructureDefinition(structure_definition) => {
64            _add_hash_tag(&mut structure_definition.meta, sha_hash)
65        }
66        Resource::CodeSystem(code_system) => _add_hash_tag(&mut code_system.meta, sha_hash),
67        Resource::ValueSet(value_set) => _add_hash_tag(&mut value_set.meta, sha_hash),
68        Resource::SearchParameter(search_parameter) => {
69            _add_hash_tag(&mut search_parameter.meta, sha_hash)
70        }
71        _ => {}
72    }
73}
74
75fn get_id(resource: &Resource) -> String {
76    match resource {
77        Resource::StructureDefinition(structure_definition) => {
78            structure_definition.id.clone().unwrap_or_default()
79        }
80        Resource::CodeSystem(code_system) => code_system.id.clone().unwrap_or_default(),
81        Resource::ValueSet(value_set) => value_set.id.clone().unwrap_or_default(),
82        Resource::SearchParameter(search_parameter) => {
83            search_parameter.id.clone().unwrap_or_default()
84        }
85        _ => todo!(
86            "Unsupported resource type '{}'",
87            resource.resource_type().as_ref()
88        ),
89    }
90}
91
92pub fn get_resource_type(resource: &Resource) -> ResourceType {
93    match resource {
94        Resource::StructureDefinition(_) => ResourceType::StructureDefinition,
95        Resource::CodeSystem(_) => ResourceType::CodeSystem,
96        Resource::ValueSet(_) => ResourceType::ValueSet,
97        Resource::SearchParameter(_) => ResourceType::SearchParameter,
98        _ => todo!(
99            "Unsupported resource type '{}'",
100            resource.resource_type().as_ref()
101        ),
102    }
103}
104
105/// This deletes existing artifacts and then reloads them. In a single transaction.
106pub async fn reset_artifacts(config: Arc<ServerConfig>) -> Result<(), OperationOutcomeError> {
107    let services = create_services(config.clone()).await?;
108
109    let transaction = services.transaction().await?;
110
111    {
112        let ctx = Arc::new(ServerCTX::system(
113            TenantId::System,
114            ProjectId::System,
115            transaction.fhir_client.clone(),
116            transaction.rate_limit.clone(),
117        ));
118
119        tracing::info!("Deleting existing CodeSystems");
120        ctx.client
121            .delete_type(
122                ctx.clone(),
123                ResourceType::CodeSystem,
124                (vec![] as Vec<(String, Vec<String>)>).into(),
125            )
126            .await?;
127        tracing::info!("Deleting existing ValueSets");
128        ctx.client
129            .delete_type(
130                ctx.clone(),
131                ResourceType::ValueSet,
132                (vec![] as Vec<(String, Vec<String>)>).into(),
133            )
134            .await?;
135        tracing::info!("Deleting existing StructureDefinitions");
136        ctx.client
137            .delete_type(
138                ctx.clone(),
139                ResourceType::StructureDefinition,
140                (vec![] as Vec<(String, Vec<String>)>).into(),
141            )
142            .await?;
143        tracing::info!("Deleting existing SearchParameters");
144        ctx.client
145            .delete_type(
146                ctx.clone(),
147                ResourceType::SearchParameter,
148                (vec![] as Vec<(String, Vec<String>)>).into(),
149            )
150            .await?;
151        _load_artifacts(ctx.clone()).await?;
152    }
153
154    transaction.commit().await?;
155
156    Ok(())
157}
158
159// Used for both reloading artifacts and reset.
160async fn _load_artifacts<Client: FHIRClient<Arc<ServerCTX<Client>>, OperationOutcomeError>>(
161    ctx: Arc<ServerCTX<Client>>,
162) -> Result<(), OperationOutcomeError> {
163    let mut hashes = HashSet::new();
164    let mut artifact_transaction_bundle_entries: Vec<BundleEntry> = vec![];
165    let mut entry_meta: Vec<(ResourceType, String, String)> = vec![];
166
167    for resource in ARTIFACT_RESOURCES.iter() {
168        let sha_hash = generate_sha256_hash(resource);
169        hashes.insert(sha_hash);
170
171        match &resource {
172            Resource::SearchParameter(_)
173            | Resource::CodeSystem(_)
174            | Resource::ValueSet(_)
175            | Resource::StructureDefinition(_) => {
176                let mut resource = resource.clone();
177                let resource_type = get_resource_type(&resource);
178                let resource_type_str = resource_type.as_ref();
179                let id = get_id(&resource);
180                let sha_hash = generate_sha256_hash(&resource);
181
182                add_hash_tag(&mut resource, sha_hash.clone());
183
184                entry_meta.push((resource_type.clone(), id.clone(), sha_hash.clone()));
185
186                artifact_transaction_bundle_entries.push(BundleEntry {
187                    resource: Some(Box::new(resource.clone())),
188                    request: Some(BundleEntryRequest {
189                        method: HttpVerb::put(),
190                        url: Box::new(
191                            format!(
192                                "{resource_type_str}?_id={id}&_tag:not={HASH_TAG_SYSTEM}|{sha_hash}"
193                            )
194                            .into(),
195                        ),
196                        ..Default::default()
197                    }),
198
199                    ..Default::default()
200                });
201            }
202            _ => {
203                // println!("Skipping resource.");
204            }
205        }
206    }
207
208    let batch_response = ctx
209        .client
210        .batch(
211            ctx.clone(),
212            Bundle {
213                type_: BundleType::batch(),
214                entry: Some(artifact_transaction_bundle_entries),
215                ..Default::default()
216            },
217        )
218        .await?;
219
220    let mut total_loaded = 0;
221    for ((resource_type, id, sha_hash), entry) in
222        entry_meta.iter().zip(batch_response.entry.iter().flatten())
223    {
224        match entry.response.as_ref().and_then(|r| r.outcome.as_deref()) {
225            Some(Resource::OperationOutcome(outcome)) if !outcome.issue.is_empty() => {
226                let issue = &outcome.issue[0];
227                let diagnostic = issue
228                    .diagnostics
229                    .as_deref()
230                    .and_then(|d| d.value.as_deref())
231                    .unwrap_or("unknown");
232
233                if issue.code == IssueType::invalid() {
234                    tracing::error!("{:#?}", outcome);
235                    panic!("INVALID");
236                } else if issue.code == IssueType::conflict() {
237                    // Ignore.
238                } else {
239                    tracing::error!(
240                        "Failed to update '{}' with id '{}'. Issue code: '{:?}', diagnostic: '{}'",
241                        resource_type.as_ref(),
242                        id,
243                        issue.code,
244                        diagnostic
245                    );
246                }
247            }
248            _ => {
249                total_loaded += 1;
250                tracing::info!(
251                    "Updated '{}' with id '{}' and sha '{}'",
252                    resource_type.as_ref(),
253                    entry
254                        .resource
255                        .as_deref()
256                        .and_then(|r| r.id().as_deref())
257                        .unwrap_or("unknown"),
258                    sha_hash.as_str()
259                );
260            }
261        }
262    }
263
264    tracing::info!(
265        "Loaded a total of '{}' artifacts with unique hashes '{}'",
266        total_loaded,
267        hashes.len()
268    );
269
270    Ok(())
271}
272
273pub async fn load_artifacts(config: Arc<ServerConfig>) -> Result<(), OperationOutcomeError> {
274    let services = create_services(config.clone()).await?;
275
276    let ctx = Arc::new(ServerCTX::system(
277        TenantId::System,
278        ProjectId::System,
279        services.fhir_client.clone(),
280        services.rate_limit.clone(),
281    ));
282
283    _load_artifacts(ctx.clone()).await
284}
285
286/// Every resource of `resource_type` in the project that matches
287/// `parameters`.
288pub async fn search_all<Repo: Repository, Search: SearchEngine>(
289    repo: &Repo,
290    search_engine: &Search,
291    tenant: &TenantId,
292    project: &ProjectId,
293    resource_type: ResourceType,
294    parameters: &[(&str, &[&str])],
295) -> Result<Vec<Resource>, OperationOutcomeError> {
296    let parameters: Vec<(String, Vec<String>)> = parameters
297        .iter()
298        .map(|(name, values)| {
299            (
300                (*name).to_string(),
301                values.iter().map(|value| (*value).to_string()).collect(),
302            )
303        })
304        .collect();
305    let results = search_engine
306        .search(
307            &SupportedFHIRVersions::R4,
308            tenant,
309            project,
310            &SearchRequest::Type(FHIRSearchTypeRequest {
311                resource_type,
312                parameters: parameters.into(),
313            }),
314            Some(SearchOptions {
315                count_limit: Some(MAX_COUNT),
316                any_of: None,
317            }),
318        )
319        .await?;
320    let version_ids: Vec<_> = results.entries.iter().map(|e| &e.version_id).collect();
321
322    repo.read_by_version_ids(tenant, project, &version_ids, CachePolicy::NoCache)
323        .await
324}
325
326/// The system project's concrete base StructureDefinitions of `kinds`.
327pub async fn get_all_sds<Repo: Repository, Search: SearchEngine>(
328    kinds: &[&str],
329    repo: &Repo,
330    search_engine: &Search,
331) -> Result<Vec<StructureDefinition>, OperationOutcomeError> {
332    let sds = search_all(
333        repo,
334        search_engine,
335        &TenantId::System,
336        &ProjectId::System,
337        ResourceType::StructureDefinition,
338        &[
339            ("kind", kinds),
340            ("abstract", &["false"]),
341            ("derivation", &["specialization"]),
342        ],
343    )
344    .await?;
345
346    Ok(sds
347        .into_iter()
348        .filter_map(|r| match r {
349            Resource::StructureDefinition(sd) => Some(sd),
350            _ => None,
351        })
352        .collect())
353}
354
355/// The system project's SearchParameters.
356pub async fn get_all_sps<Repo: Repository, Search: SearchEngine>(
357    repo: &Repo,
358    search_engine: &Search,
359) -> Result<Vec<SearchParameter>, OperationOutcomeError> {
360    let sps = search_all(
361        repo,
362        search_engine,
363        &TenantId::System,
364        &ProjectId::System,
365        ResourceType::SearchParameter,
366        &[],
367    )
368    .await?;
369
370    Ok(sps
371        .into_iter()
372        .filter_map(|r| match r {
373            Resource::SearchParameter(sp) => Some(sp),
374            _ => None,
375        })
376        .collect())
377}
378
379/// Every `OperationDefinition` installed for the user.
380/// Includes the embeded operations we create
381pub async fn get_all_operation_definitions<
382    Client: FHIRClient<Arc<ServerCTX<Client>>, OperationOutcomeError> + 'static,
383>(
384    ctx: Arc<ServerCTX<Client>>,
385) -> Result<Vec<OperationDefinition>, OperationOutcomeError> {
386    let operations = ctx
387        .client
388        .search_type(
389            ctx.clone(),
390            ResourceType::OperationDefinition,
391            (vec![] as Vec<(String, Vec<String>)>).into(),
392        )
393        .await?
394        .entry
395        .unwrap_or_default()
396        .into_iter()
397        .filter_map(|e| e.resource)
398        .filter_map(|r| match *r {
399            Resource::OperationDefinition(operation) => Some(operation),
400            _ => None,
401        })
402        .collect();
403
404    Ok(operations)
405}
haste_server/load_artifacts.rs

haste_server/
load_artifacts.rs
