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/// The watermark half of a sequence poll; see
17/// [`ResourceSequential::max_safe_sequence`]. Its own statement on purpose:
18/// `max_safe_seq` reads the sequence and scans `pg_locks` inside the
19/// function, and a row select that uses the result must take its snapshot
20/// after that scan, which a later statement under READ COMMITTED does.
21async fn max_safe_sequence_helper(
22    executor: &mut PgConnection,
23) -> Result<i64, OperationOutcomeError> {
24    let safe_sequence: Option<i64> =
25        sqlx::query_scalar("SELECT max_safe_seq('resources_sequence_seq')")
26            .fetch_one(&mut *executor)
27            .await
28            .map_err(StoreError::from)?;
29
30    Ok(safe_sequence.unwrap_or(0))
31}
32
33// 1. Concrete helper function accepting an explicit reference to remove HRTB issues entirely
34async fn get_sequence_helper(
35    executor: &mut PgConnection,
36    tenant_id: &TenantId,
37    project_id: &ProjectId,
38    cur_sequence: u64,
39    safe_sequence: i64,
40    count: Option<u64>,
41) -> Result<Vec<ResourcePollingValue>, OperationOutcomeError> {
42    let result = sqlx::query_as::<
43        _,
44        (
45            String,
46            TenantId,
47            ProjectId,
48            String,
49            String,
50            FHIRMethod,
51            i64,
52            FHIRJson<Resource>,
53        ),
54    >(
55        r"
56            SELECT id, tenant, project, version_id, resource_type, fhir_method, sequence, resource
57            FROM resources
58            WHERE tenant = $1 AND project = $2 AND sequence > $3 AND sequence <= $4
59            ORDER BY sequence
60            LIMIT $5
61        ",
62    )
63    .bind(tenant_id.as_ref())
64    .bind(project_id.as_ref())
65    .bind(cur_sequence.cast_signed())
66    .bind(safe_sequence)
67    .bind(count.unwrap_or(100).cast_signed())
68    .fetch_all(executor)
69    .await
70    .map_err(StoreError::from)?;
71
72    result
73        .into_iter()
74        .map(
75            |(
76                id,
77                tenant,
78                project,
79                version_id,
80                resource_type_str,
81                fhir_method,
82                sequence,
83                resource,
84            )| {
85                let resource_type = ResourceType::try_from(resource_type_str).map_err(|_| {
86                    OperationOutcomeError::error(
87                        IssueType::structure(),
88                        "Invalid resource type encountered during sequence polling.".to_string(),
89                    )
90                })?;
91
92                Ok::<ResourcePollingValue, OperationOutcomeError>(ResourcePollingValue {
93                    id: ResourceId::new(id),
94                    tenant,
95                    project,
96                    version_id,
97                    resource_type,
98                    fhir_method,
99                    sequence,
100                    resource,
101                })
102            },
103        )
104        .collect()
105}
106
107// 2. Trait implementation matching your PGConnection enum
108impl ResourceSequential for PGConnection {
109    async fn max_safe_sequence(&self) -> Result<i64, OperationOutcomeError> {
110        match self {
111            PGConnection::Pool(pool, _) => {
112                let mut conn = pool.acquire().await.map_err(StoreError::from)?;
113                max_safe_sequence_helper(&mut conn).await
114            }
115            PGConnection::Transaction(tx, _, _) => {
116                let mut conn = tx.lock().await;
117                max_safe_sequence_helper(&mut conn).await
118            }
119        }
120    }
121
122    async fn get_sequence(
123        &self,
124        tenant_id: &TenantId,
125        project_id: &ProjectId,
126        sequence_id: u64,
127        safe_sequence: i64,
128        count: Option<u64>,
129    ) -> Result<Vec<ResourcePollingValue>, OperationOutcomeError> {
130        match self {
131            PGConnection::Pool(pool, _) => {
132                let mut conn = pool.acquire().await.map_err(StoreError::from)?;
133                get_sequence_helper(
134                    &mut conn,
135                    tenant_id,
136                    project_id,
137                    sequence_id,
138                    safe_sequence,
139                    count,
140                )
141                .await
142            }
143            PGConnection::Transaction(tx, _, _) => {
144                let mut conn = tx.lock().await;
145                // Pass the mutable reference to the underlying PgConnection handle
146                get_sequence_helper(
147                    &mut conn,
148                    tenant_id,
149                    project_id,
150                    sequence_id,
151                    safe_sequence,
152                    count,
153                )
154                .await
155            }
156        }
157    }
158}
haste_repository/pg/sequence.rs

haste_repository/pg/
sequence.rs
