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

haste_server/
load_artifacts.rs
