haste_repository/pg/
fhir.rs

1use crate::{
2    fhir::{CachePolicy, FHIRRepository, ResourceHistoryValue},
3    pg::{
4        PGConnection, StoreError,
5        utilities::{commit_transaction, create_transaction},
6    },
7    types::{FHIRMethod, SupportedFHIRVersions},
8    utilities,
9};
10use haste_fhir_client::{
11    request::HistoryRequest,
12    url::{ParsedParameter, ParsedParameters},
13};
14use haste_fhir_model::r4::{
15    datetime::parse_datetime,
16    generated::{
17        resources::{Resource, ResourceType},
18        terminology::IssueType,
19    },
20    sqlx::{FHIRJson, FHIRJsonRef},
21};
22use haste_fhir_operation_error::OperationOutcomeError;
23use haste_jwt::{ProjectId, ResourceId, TenantId, VersionId, claims::UserTokenClaims};
24use moka::future::Cache;
25use sqlx::{PgExecutor, Postgres, QueryBuilder, query_builder::Separated};
26use std::sync::Arc;
27use tokio::sync::Mutex;
28
29#[derive(sqlx::FromRow, Debug)]
30struct ReturnVersionedResource {
31    resource: FHIRJson<Resource>,
32    version_id: VersionId,
33}
34
35#[derive(sqlx::FromRow, Debug)]
36struct HistoryValue {
37    pub resource: FHIRJson<Resource>,
38    pub request_method: String,
39}
40
41async fn read_version_ids_from_cache<'a>(
42    cache: &Cache<VersionId, Resource>,
43    version_ids: &'a [&VersionId],
44) -> (Vec<Resource>, Vec<&'a VersionId>) {
45    let mut remaining_version_ids = vec![];
46    let mut cached_resources = vec![];
47    for version_id in version_ids {
48        if let Some(resource) = cache.get(*version_id).await {
49            cached_resources.push(resource);
50        } else {
51            remaining_version_ids.push(*version_id);
52        }
53    }
54
55    (cached_resources, remaining_version_ids)
56}
57
58impl FHIRRepository for PGConnection {
59    async fn create(
60        &self,
61        tenant: &TenantId,
62        project: &ProjectId,
63        author: &UserTokenClaims,
64        fhir_version: &SupportedFHIRVersions,
65        resource: &mut Resource,
66    ) -> Result<Resource, OperationOutcomeError> {
67        match &self {
68            PGConnection::Pool(_pool, _) => {
69                let tx = create_transaction(self, true).await?;
70                let res = {
71                    let mut conn = tx.lock().await;
72                    create(&mut **conn, tenant, project, author, fhir_version, resource).await?
73                };
74                commit_transaction(tx).await?;
75                Ok(res)
76            }
77            PGConnection::Transaction(tx, _) => {
78                let mut tx = tx.lock().await;
79                create(&mut **tx, tenant, project, author, fhir_version, resource).await
80            }
81        }
82    }
83
84    async fn delete(
85        &self,
86        tenant: &TenantId,
87        project: &ProjectId,
88        author: &UserTokenClaims,
89        fhir_version: &SupportedFHIRVersions,
90        resource: &mut Resource,
91        id: &str,
92    ) -> Result<Resource, OperationOutcomeError> {
93        match self {
94            PGConnection::Pool(_pool, _) => {
95                let tx = create_transaction(self, true).await?;
96                let res = {
97                    let mut conn = tx.lock().await;
98                    delete(
99                        &mut **conn,
100                        tenant,
101                        project,
102                        author,
103                        fhir_version,
104                        resource,
105                        id,
106                    )
107                    .await?
108                };
109                commit_transaction(tx).await?;
110                Ok(res)
111            }
112            PGConnection::Transaction(tx, _) => {
113                let mut conn = tx.lock().await;
114                // Handle PgConnection connection
115                delete(
116                    &mut **conn,
117                    tenant,
118                    project,
119                    author,
120                    fhir_version,
121                    resource,
122                    id,
123                )
124                .await
125            }
126        }
127    }
128
129    async fn update(
130        &self,
131        tenant: &TenantId,
132        project: &ProjectId,
133        author: &UserTokenClaims,
134        fhir_version: &SupportedFHIRVersions,
135        resource: &mut Resource,
136        id: &str,
137    ) -> Result<Resource, OperationOutcomeError> {
138        match self {
139            PGConnection::Pool(_pool, _) => {
140                let tx = create_transaction(self, true).await?;
141                let res = {
142                    let mut conn = tx.lock().await;
143                    update(
144                        &mut **conn,
145                        tenant,
146                        project,
147                        author,
148                        fhir_version,
149                        resource,
150                        id,
151                    )
152                    .await?
153                };
154
155                commit_transaction(tx).await?;
156                Ok(res)
157            }
158            PGConnection::Transaction(tx, _) => {
159                let mut conn = tx.lock().await;
160                // Handle PgConnection connection
161                update(
162                    &mut **conn,
163                    tenant,
164                    project,
165                    author,
166                    fhir_version,
167                    resource,
168                    id,
169                )
170                .await
171            }
172        }
173    }
174
175    async fn read_by_version_ids(
176        &self,
177        tenant_id: &TenantId,
178        project_id: &ProjectId,
179        version_ids: &[&VersionId],
180        cache_policy: CachePolicy,
181    ) -> Result<Vec<Resource>, OperationOutcomeError> {
182        if version_ids.is_empty() {
183            return Ok(Vec::new());
184        }
185
186        let (cached_result, remaining_version_ids) =
187            read_version_ids_from_cache(self.cache(), version_ids).await;
188
189        if remaining_version_ids.is_empty() {
190            return Ok(cached_result);
191        }
192
193        match self {
194            PGConnection::Pool(pool, cache) => {
195                let res = read_by_version_ids(pool, tenant_id, project_id, &remaining_version_ids)
196                    .await?;
197
198                if cache_policy == CachePolicy::Cache {
199                    for v in &res {
200                        cache
201                            .insert(v.version_id.clone(), v.resource.0.clone())
202                            .await;
203                    }
204                }
205
206                Ok(cached_result
207                    .into_iter()
208                    .chain(res.into_iter().map(|r| r.resource.0))
209                    .collect::<Vec<_>>())
210            }
211            PGConnection::Transaction(tx, cache) => {
212                let mut conn = tx.lock().await;
213                // Handle PgConnection connection
214                let res =
215                    read_by_version_ids(&mut **conn, tenant_id, project_id, &remaining_version_ids)
216                        .await?;
217
218                if cache_policy == CachePolicy::Cache {
219                    for v in &res {
220                        cache
221                            .insert(v.version_id.clone(), v.resource.0.clone())
222                            .await;
223                    }
224                }
225
226                Ok(cached_result
227                    .into_iter()
228                    .chain(res.into_iter().map(|r| r.resource.0))
229                    .collect::<Vec<_>>())
230            }
231        }
232    }
233
234    async fn read_latest(
235        &self,
236        tenant_id: &TenantId,
237        project_id: &ProjectId,
238        resource_type: &ResourceType,
239        resource_id: &ResourceId,
240    ) -> Result<Option<Resource>, OperationOutcomeError> {
241        match self {
242            PGConnection::Pool(pool, _) => {
243                let res =
244                    read_latest(pool, tenant_id, project_id, resource_type, resource_id).await?;
245                Ok(res)
246            }
247            PGConnection::Transaction(tx, _) => {
248                let mut conn = tx.lock().await;
249                // Handle PgConnection connection
250                read_latest(
251                    &mut **conn,
252                    tenant_id,
253                    project_id,
254                    resource_type,
255                    resource_id,
256                )
257                .await
258            }
259        }
260    }
261
262    async fn history(
263        &self,
264        tenant_id: &TenantId,
265        project_id: &ProjectId,
266        request: &HistoryRequest,
267    ) -> Result<Vec<ResourceHistoryValue>, OperationOutcomeError> {
268        match self {
269            PGConnection::Pool(pool, _) => history(pool, tenant_id, project_id, request).await,
270            PGConnection::Transaction(tx, _) => {
271                let mut conn = tx.lock().await;
272                // Handle PgConnection connection
273                history(&mut **conn, tenant_id, project_id, request).await
274            }
275        }
276    }
277
278    fn in_transaction(&self) -> bool {
279        matches!(self, PGConnection::Transaction(_tx, _))
280    }
281
282    async fn transaction(&self, is_updating_sequence: bool) -> Result<Self, OperationOutcomeError> {
283        let tx = create_transaction(self, is_updating_sequence).await?;
284        Ok(PGConnection::Transaction(tx, self.cache().clone()))
285    }
286
287    async fn commit(self) -> Result<(), OperationOutcomeError> {
288        match self {
289            PGConnection::Pool(_pool, _) => Err(StoreError::NotTransaction.into()),
290            PGConnection::Transaction(tx, _) => commit_transaction(tx).await,
291        }
292    }
293
294    async fn rollback(self) -> Result<(), OperationOutcomeError> {
295        match self {
296            PGConnection::Pool(_pool, _) => Err(StoreError::NotTransaction.into()),
297            PGConnection::Transaction(tx, _) => {
298                let conn = Mutex::into_inner(
299                    Arc::try_unwrap(tx).map_err(|_e| StoreError::FailedCommitTransaction)?,
300                );
301
302                // Handle PgConnection connection
303                conn.rollback().await.map_err(StoreError::from)?;
304                Ok(())
305            }
306        }
307    }
308}
309
310async fn create<'a, 'e, E>(
311    executor: E,
312    tenant: &'a TenantId,
313    project: &'a ProjectId,
314    author: &'a UserTokenClaims,
315    fhir_version: &'a SupportedFHIRVersions,
316    resource: &'a mut Resource,
317) -> Result<Resource, OperationOutcomeError>
318where
319    E: PgExecutor<'e>,
320{
321    utilities::set_resource_id(resource, None)?;
322    utilities::set_version_id(resource)?;
323
324    let result = sqlx::query_as::<_, (FHIRJson<Resource>,)>(
325        r"
326            INSERT INTO resources (tenant, project, author_id, fhir_version, resource, deleted, request_method, author_type, fhir_method)
327            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
328            RETURNING resource
329        ",
330    )
331    .bind(tenant.as_ref())
332    .bind(project.as_ref())
333    .bind(author.sub.as_ref())
334    .bind(fhir_version)
335    .bind(FHIRJsonRef(resource))
336    .bind(false)
337    .bind("POST")
338    .bind(author.resource_type.as_ref())
339    .bind(FHIRMethod::Create)
340    .fetch_one(executor)
341    .await
342    .map_err(StoreError::from)?;
343
344    Ok(result.0.0)
345}
346
347async fn delete<'a, 'e, E>(
348    executor: E,
349    tenant: &'a TenantId,
350    project: &'a ProjectId,
351    author: &'a UserTokenClaims,
352    fhir_version: &'a SupportedFHIRVersions,
353    resource: &'a mut Resource,
354    id: &'a str,
355) -> Result<Resource, OperationOutcomeError>
356where
357    E: PgExecutor<'e>,
358{
359    utilities::set_resource_id(resource, Some(id.to_string()))?;
360    utilities::set_version_id(resource)?;
361
362    let result = sqlx::query_as::<_, (FHIRJson<Resource>,)>(
363        r"
364            INSERT INTO resources (tenant, project, author_id, fhir_version, resource, deleted, request_method, author_type, fhir_method)
365            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
366            RETURNING resource
367        ",
368    )
369    .bind(tenant.as_ref())
370    .bind(project.as_ref())
371    .bind(author.sub.as_ref())
372    .bind(fhir_version)
373    .bind(FHIRJsonRef(resource))
374    .bind(true)
375    .bind("DELETE")
376    .bind(author.resource_type.as_ref())
377    .bind(FHIRMethod::Delete)
378    .fetch_one(executor)
379    .await
380    .map_err(StoreError::from)?;
381
382    Ok(result.0.0)
383}
384
385async fn update<'a, 'e, E>(
386    executor: E,
387    tenant: &'a TenantId,
388    project: &'a ProjectId,
389    author: &'a UserTokenClaims,
390    fhir_version: &'a SupportedFHIRVersions,
391    resource: &'a mut Resource,
392    id: &'a str,
393) -> Result<Resource, OperationOutcomeError>
394where
395    E: PgExecutor<'e>,
396{
397    utilities::set_resource_id(resource, Some(id.to_string()))?;
398    utilities::set_version_id(resource)?;
399
400    let result = sqlx::query_as::<_, (FHIRJson<Resource>,)>(
401        r"
402            INSERT INTO resources (tenant, project, author_id, fhir_version, resource, deleted, request_method, author_type, fhir_method)
403            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
404            RETURNING resource
405        ",
406    )
407    .bind(tenant.as_ref())
408    .bind(project.as_ref())
409    .bind(author.sub.as_ref())
410    .bind(fhir_version)
411    .bind(FHIRJsonRef(resource))
412    .bind(false)
413    .bind("PUT")
414    .bind(author.resource_type.as_ref())
415    .bind(FHIRMethod::Update)
416    .fetch_one(executor)
417    .await
418    .map_err(StoreError::from)?;
419
420    Ok(result.0.0)
421}
422
423async fn read_by_version_ids<'a, 'e, E>(
424    executor: E,
425    tenant_id: &'a TenantId,
426    project_id: &'a ProjectId,
427    version_ids: &'a Vec<&'a VersionId>,
428) -> Result<Vec<ReturnVersionedResource>, OperationOutcomeError>
429where
430    E: PgExecutor<'e>,
431{
432    let mut query_builder: QueryBuilder<sqlx::Postgres> =
433        QueryBuilder::new("SELECT resource, version_id FROM resources WHERE tenant = ");
434
435    query_builder
436        .push_bind(tenant_id.as_ref())
437        .push(" AND project =")
438        .push_bind(project_id.as_ref());
439
440    query_builder.push(" AND version_id in (");
441
442    let mut separated = query_builder.separated(", ");
443    for version_id in version_ids {
444        separated.push_bind(version_id.as_ref());
445    }
446    separated.push_unseparated(")");
447
448    query_builder.push(" ORDER BY array_position(array[");
449    let mut order_separator = query_builder.separated(", ");
450    for version_id in version_ids {
451        order_separator.push_bind(version_id.as_ref());
452    }
453    query_builder.push("], version_id)");
454
455    let query = query_builder.build_query_as::<ReturnVersionedResource>();
456
457    let response: Vec<ReturnVersionedResource> =
458        query.fetch_all(executor).await.map_err(StoreError::from)?;
459
460    Ok(response)
461}
462
463async fn read_latest<'a, 'e, E>(
464    executor: E,
465    tenant_id: &'a TenantId,
466    project_id: &'a ProjectId,
467    resource_type: &'a ResourceType,
468    resource_id: &'a ResourceId,
469) -> Result<Option<Resource>, OperationOutcomeError>
470where
471    E: PgExecutor<'e>,
472{
473    let response = sqlx::query_as::<_, (FHIRJson<Resource>, bool)>(
474        r"
475            SELECT resource, deleted
476            FROM resources
477            WHERE tenant = $1 AND project = $2 AND id = $3 AND resource_type = $4
478            ORDER BY sequence DESC
479            LIMIT 1
480        ",
481    )
482    .bind(tenant_id.as_ref())
483    .bind(project_id.as_ref())
484    .bind(resource_id.as_ref())
485    .bind(resource_type.as_ref())
486    .fetch_optional(executor)
487    .await
488    .map_err(StoreError::from)?;
489
490    match response {
491        Some((_, true)) => Ok(None),
492        Some((json, _)) => Ok(Some(json.0)),
493        None => Ok(None),
494    }
495}
496
497fn process_history_parameters<'a>(
498    parameters: &'a ParsedParameters,
499    clauses: &mut Separated<'_, 'a, Postgres, &str>,
500) -> Result<(), OperationOutcomeError> {
501    for parameter in parameters.parameters() {
502        match parameter {
503            ParsedParameter::Result(result_param) => {
504                if result_param.name.as_str() == "_since" {
505                    if let Some(value) = result_param.value.first() {
506                        let date_time = parse_datetime(value.as_str()).map_err(|e| {
507                            OperationOutcomeError::fatal(
508                                IssueType::invalid(),
509                                format!("Invalid _since parameter datetime: {e:?}"),
510                            )
511                        })?;
512
513                        clauses.push(" created_at >= ").push_bind_unseparated(
514                            chrono::DateTime::try_from(date_time).map_err(|e| {
515                                OperationOutcomeError::fatal(
516                                    IssueType::invalid(),
517                                    format!("Invalid _since parameter datetime: {e:?}"),
518                                )
519                            })?,
520                        );
521                    }
522                } else {
523                    // Ignore offset and count parameter as these parameters are held separately and not used in the where clause.
524                }
525            }
526            ParsedParameter::Resource(_) => {
527                return Err(OperationOutcomeError::fatal(
528                    IssueType::not_supported(),
529                    format!(
530                        "Parameter '{}' is not supported for history requests.",
531                        parameter.name()
532                    ),
533                ));
534            }
535        }
536    }
537
538    Ok(())
539}
540
541async fn history<'a, 'e, E>(
542    executor: E,
543    tenant: &'a TenantId,
544    project: &'a ProjectId,
545    history_request: &'a HistoryRequest,
546) -> Result<Vec<ResourceHistoryValue>, OperationOutcomeError>
547where
548    E: PgExecutor<'e>,
549{
550    let mut query_builder: QueryBuilder<sqlx::Postgres> =
551        QueryBuilder::new(r"SELECT resource, request_method FROM resources WHERE ");
552
553    let mut clauses = query_builder.separated(" AND ");
554    clauses
555        .push(" tenant = ")
556        .push_bind_unseparated(tenant.as_ref())
557        .push(" project = ")
558        .push_bind_unseparated(project.as_ref());
559
560    let history_parameters = match history_request {
561        HistoryRequest::Instance(history_instance_request) => &history_instance_request.parameters,
562        HistoryRequest::Type(history_type_request) => &history_type_request.parameters,
563        HistoryRequest::System(system_request) => &system_request.parameters,
564    };
565
566    process_history_parameters(history_parameters, &mut clauses)?;
567
568    match history_request {
569        HistoryRequest::Instance(history_instance_request) => {
570            clauses
571                .push(" resource_type = ")
572                .push_bind_unseparated(history_instance_request.resource_type.as_ref())
573                .push(" id = ")
574                .push_bind_unseparated(&history_instance_request.id);
575        }
576        HistoryRequest::Type(history_type_request) => {
577            clauses
578                .push(" resource_type = ")
579                .push_bind_unseparated(history_type_request.resource_type.as_ref());
580        }
581        HistoryRequest::System(_request) => {}
582    }
583
584    let limit = if let Some(ParsedParameter::Result(count_param)) = history_parameters.get("_count")
585    {
586        std::cmp::min(
587            1000,
588            count_param
589                .value
590                .first()
591                .and_then(|v| v.parse::<i64>().ok())
592                .unwrap_or(100),
593        )
594    } else {
595        1000
596    };
597
598    if limit < 0 {
599        return Err(OperationOutcomeError::fatal(
600            IssueType::invalid(),
601            "Invalid _count parameter value. Must be greater than or equal to 0.".to_string(),
602        ));
603    }
604
605    query_builder
606        .push(" ORDER BY sequence DESC LIMIT ")
607        .push_bind(limit);
608
609    if let Some(ParsedParameter::Result(offset_param)) = history_parameters.get("_offset") {
610        let offset = offset_param
611            .value
612            .first()
613            .and_then(|v| v.parse::<i64>().ok())
614            .unwrap_or(0);
615
616        if offset < 0 {
617            return Err(OperationOutcomeError::fatal(
618                IssueType::invalid(),
619                "Invalid _offset parameter value. Must be greater than or equal to 0.".to_string(),
620            ));
621        }
622
623        query_builder.push(" OFFSET ").push_bind(offset);
624    }
625
626    let query = query_builder.build_query_as::<HistoryValue>();
627
628    let result: Vec<HistoryValue> = query.fetch_all(executor).await.map_err(StoreError::from)?;
629
630    Ok(result
631        .into_iter()
632        .map(|r| ResourceHistoryValue {
633            resource: r.resource.0,
634            request_method: r.request_method,
635        })
636        .collect::<Vec<_>>())
637}
haste_repository/pg/fhir.rs

haste_repository/pg/
fhir.rs
