haste_repository/pg/
sequence.rs

1use haste_fhir_model::r4::{
2    generated::resources::{Resource, ResourceType},
3    generated::terminology::IssueType,
4    sqlx::FHIRJson,
5};
6use haste_fhir_operation_error::OperationOutcomeError;
7use haste_jwt::{ProjectId, ResourceId, TenantId};
8use sqlx::PgConnection;
9
10use crate::{
11    pg::{PGConnection, StoreError},
12    sequence::{ResourcePollingValue, ResourceSequential},
13    types::FHIRMethod,
14};
15
16// 1. Concrete helper function accepting an explicit reference to remove HRTB issues entirely
17async fn get_sequence_helper(
18    executor: &mut PgConnection,
19    tenant_id: &TenantId,
20    cur_sequence: u64,
21    count: Option<u64>,
22) -> Result<Vec<ResourcePollingValue>, OperationOutcomeError> {
23    let safe_sequence_row = sqlx::query_as::<_, (Option<i64>,)>(
24        "SELECT max_safe_seq('resources_sequence_seq') as max_safe_seq",
25    )
26    .fetch_one(&mut *executor)
27    .await
28    .map_err(StoreError::from)?;
29
30    let safe_sequence = safe_sequence_row.0.unwrap_or(0);
31
32    let result = sqlx::query_as::<
33        _,
34        (
35            String,
36            TenantId,
37            ProjectId,
38            String,
39            String,
40            FHIRMethod,
41            i64,
42            FHIRJson<Resource>,
43        ),
44    >(
45        r"
46            SELECT id, tenant, project, version_id, resource_type, fhir_method, sequence, resource
47            FROM resources
48            WHERE tenant = $1 AND sequence > $2 AND sequence <= $3
49            ORDER BY sequence
50            LIMIT $4
51        ",
52    )
53    .bind(tenant_id.as_ref())
54    .bind(cur_sequence.cast_signed())
55    .bind(safe_sequence)
56    .bind(count.unwrap_or(100).cast_signed())
57    .fetch_all(executor)
58    .await
59    .map_err(StoreError::from)?;
60
61    result
62        .into_iter()
63        .map(
64            |(
65                id,
66                tenant,
67                project,
68                version_id,
69                resource_type_str,
70                fhir_method,
71                sequence,
72                resource,
73            )| {
74                let resource_type = ResourceType::try_from(resource_type_str).map_err(|_| {
75                    OperationOutcomeError::error(
76                        IssueType::structure(),
77                        "Invalid resource type encountered during sequence polling.".to_string(),
78                    )
79                })?;
80
81                Ok::<ResourcePollingValue, OperationOutcomeError>(ResourcePollingValue {
82                    id: ResourceId::new(id),
83                    tenant,
84                    project,
85                    version_id,
86                    resource_type,
87                    fhir_method,
88                    sequence,
89                    resource,
90                })
91            },
92        )
93        .collect()
94}
95
96// 2. Trait implementation matching your PGConnection enum
97impl ResourceSequential for PGConnection {
98    async fn get_sequence(
99        &self,
100        tenant_id: &TenantId,
101        sequence_id: u64,
102        count: Option<u64>,
103    ) -> Result<Vec<ResourcePollingValue>, OperationOutcomeError> {
104        match self {
105            PGConnection::Pool(pool, _) => {
106                // Acquire a dedicated connection from the pool so both queries execute
107                // sequentially on the same PgConnection, matching the transaction path.
108                let mut conn = pool.acquire().await.map_err(StoreError::from)?;
109                get_sequence_helper(&mut conn, tenant_id, sequence_id, count).await
110            }
111            PGConnection::Transaction(tx, _) => {
112                let mut conn = tx.lock().await;
113                // Pass the mutable reference to the underlying PgConnection handle
114                get_sequence_helper(&mut conn, tenant_id, sequence_id, count).await
115            }
116        }
117    }
118}
haste_repository/pg/sequence.rs

haste_repository/pg/
sequence.rs
