haste_repository/pg/
pending.rs

1//! Writing rows to the `resources` table for `PGConnection`.
2//!
3//! A `PGConnection::Pool` write has nothing to buffer — it inserts
4//! immediately via [`insert`], which only ever borrows the resource, so it
5//! never clones or allocates for it. A `PGConnection::Transaction` write is
6//! queued in [`PendingRows`] instead: earlier writes on the same transaction
7//! must stay invisible to Postgres until flush/commit, and reads on that
8//! transaction call [`PendingRows::flush`] first so they observe them. Since
9//! a queued row has to outlive the call that created it, it needs its own
10//! owned copy of the resource.
11//!
12use crate::{
13    pg::StoreError,
14    types::{FHIRMethod, SupportedFHIRVersions},
15};
16use haste_fhir_model::r4::{generated::resources::Resource, sqlx::FHIRJsonRef};
17use haste_fhir_operation_error::OperationOutcomeError;
18use haste_jwt::{AuthorId, AuthorKind, ProjectId, TenantId, claims::UserTokenClaims};
19use sqlx::{PgExecutor, Postgres, QueryBuilder};
20use std::sync::Arc;
21use tokio::sync::Mutex;
22
23/// Common field access for anything that can be written as a `resources`
24/// row, whether it owns its data ([`PendingResourceRow`]) or only borrows it
25/// ([`BorrowedResourceRow`]). Lets `insert_batch` bind either kind without
26/// caring which.
27trait ResourceRowFields {
28    fn tenant(&self) -> &TenantId;
29    fn project(&self) -> &ProjectId;
30    fn author_id(&self) -> &AuthorId;
31    fn author_type(&self) -> &AuthorKind;
32    fn fhir_version(&self) -> &SupportedFHIRVersions;
33    fn resource(&self) -> &Resource;
34    fn deleted(&self) -> bool;
35    fn request_method(&self) -> &str;
36    fn fhir_method(&self) -> &FHIRMethod;
37}
38
39/// A single buffered `resources` row awaiting a batched multi-row INSERT.
40#[derive(Debug, Clone)]
41struct PendingResourceRow {
42    tenant: TenantId,
43    project: ProjectId,
44    author_id: AuthorId,
45    author_type: AuthorKind,
46    fhir_version: SupportedFHIRVersions,
47    resource: Resource,
48    deleted: bool,
49    request_method: &'static str,
50    fhir_method: FHIRMethod,
51}
52
53impl ResourceRowFields for PendingResourceRow {
54    fn tenant(&self) -> &TenantId {
55        &self.tenant
56    }
57    fn project(&self) -> &ProjectId {
58        &self.project
59    }
60    fn author_id(&self) -> &AuthorId {
61        &self.author_id
62    }
63    fn author_type(&self) -> &AuthorKind {
64        &self.author_type
65    }
66    fn fhir_version(&self) -> &SupportedFHIRVersions {
67        &self.fhir_version
68    }
69    fn resource(&self) -> &Resource {
70        &self.resource
71    }
72    fn deleted(&self) -> bool {
73        self.deleted
74    }
75    fn request_method(&self) -> &str {
76        self.request_method
77    }
78    fn fhir_method(&self) -> &FHIRMethod {
79        &self.fhir_method
80    }
81}
82
83/// A `resources` row for an immediate, unbuffered INSERT — every field is
84/// borrowed straight from the caller, so writing it costs no clone and no
85/// allocation beyond the query itself.
86struct BorrowedResourceRow<'a> {
87    tenant: &'a TenantId,
88    project: &'a ProjectId,
89    author_id: &'a AuthorId,
90    author_type: &'a AuthorKind,
91    fhir_version: &'a SupportedFHIRVersions,
92    resource: &'a Resource,
93    deleted: bool,
94    request_method: &'static str,
95    fhir_method: FHIRMethod,
96}
97
98impl ResourceRowFields for BorrowedResourceRow<'_> {
99    fn tenant(&self) -> &TenantId {
100        self.tenant
101    }
102    fn project(&self) -> &ProjectId {
103        self.project
104    }
105    fn author_id(&self) -> &AuthorId {
106        self.author_id
107    }
108    fn author_type(&self) -> &AuthorKind {
109        self.author_type
110    }
111    fn fhir_version(&self) -> &SupportedFHIRVersions {
112        self.fhir_version
113    }
114    fn resource(&self) -> &Resource {
115        self.resource
116    }
117    fn deleted(&self) -> bool {
118        self.deleted
119    }
120    fn request_method(&self) -> &str {
121        self.request_method
122    }
123    fn fhir_method(&self) -> &FHIRMethod {
124        &self.fhir_method
125    }
126}
127
128/// Inserts a single `resources` row immediately. Used for
129/// `PGConnection::Pool` writes, which are never buffered and so never need
130/// to own the resource.
131#[allow(clippy::too_many_arguments)]
132pub async fn execute<'e, E>(
133    executor: E,
134    tenant: &TenantId,
135    project: &ProjectId,
136    author: &UserTokenClaims,
137    fhir_version: &SupportedFHIRVersions,
138    resource: &Resource,
139    deleted: bool,
140    request_method: &'static str,
141    fhir_method: FHIRMethod,
142) -> Result<(), OperationOutcomeError>
143where
144    E: PgExecutor<'e>,
145{
146    let row = BorrowedResourceRow {
147        tenant,
148        project,
149        author_id: &author.sub,
150        author_type: &author.resource_type,
151        fhir_version,
152        resource,
153        deleted,
154        request_method,
155        fhir_method,
156    };
157
158    insert_resource_updates(executor, std::slice::from_ref(&row)).await
159}
160
161/// Rows queued on an open transaction for one batched multi-row INSERT at
162/// flush/commit time, instead of one INSERT per write. Cheap to clone — it
163/// shares the same underlying buffer, which matters when a nested
164/// `transaction()` call reuses its parent's queue.
165#[derive(Debug, Clone, Default)]
166pub struct PendingRows(Arc<Mutex<Vec<PendingResourceRow>>>);
167
168impl PendingRows {
169    #[must_use]
170    pub fn new() -> Self {
171        Self::default()
172    }
173
174    /// Buffers a row. Unlike [`insert`], `resource` must be an owned copy —
175    /// this data has to outlive the call that queued it, until `flush` runs.
176    #[allow(clippy::too_many_arguments)]
177    pub async fn push(
178        &self,
179        tenant: &TenantId,
180        project: &ProjectId,
181        author: &UserTokenClaims,
182        fhir_version: &SupportedFHIRVersions,
183        resource: Resource,
184        deleted: bool,
185        request_method: &'static str,
186        fhir_method: FHIRMethod,
187    ) {
188        self.0.lock().await.push(PendingResourceRow {
189            tenant: tenant.clone(),
190            project: project.clone(),
191            author_id: author.sub.clone(),
192            author_type: author.resource_type.clone(),
193            fhir_version: fhir_version.clone(),
194            resource,
195            deleted,
196            request_method,
197            fhir_method,
198        });
199    }
200
201    /// Drains and inserts every buffered row, so a subsequent read on the
202    /// same transaction observes prior writes that haven't reached Postgres
203    /// yet. The internal lock is released before tx is locked, so the two
204    /// are never held simultaneously.
205    ///
206    /// # Errors
207    ///
208    /// Returns an [`OperationOutcomeError`] if inserting the buffered rows
209    /// into Postgres fails.
210    pub async fn flush(
211        &self,
212        tx: &Arc<Mutex<sqlx::Transaction<'static, Postgres>>>,
213    ) -> Result<(), OperationOutcomeError> {
214        let rows = {
215            let mut guard = self.0.lock().await;
216            std::mem::take(&mut *guard)
217        };
218
219        if rows.is_empty() {
220            return Ok(());
221        }
222
223        let mut conn = tx.lock().await;
224        insert_resource_updates(&mut **conn, &rows).await
225    }
226}
227
228/// Executes one multi-row INSERT for every row given. No-op on an empty
229/// slice (`QueryBuilder::push_values` panics if given zero tuples).
230async fn insert_resource_updates<'e, E, R>(
231    executor: E,
232    rows: &[R],
233) -> Result<(), OperationOutcomeError>
234where
235    E: PgExecutor<'e>,
236    R: ResourceRowFields,
237{
238    if rows.is_empty() {
239        return Ok(());
240    }
241
242    let mut query_builder: QueryBuilder<Postgres> = QueryBuilder::new(
243        "INSERT INTO resources (tenant, project, author_id, fhir_version, resource, deleted, request_method, author_type, fhir_method) ",
244    );
245
246    query_builder.push_values(rows, |mut b, row| {
247        b.push_bind(row.tenant().as_ref())
248            .push_bind(row.project().as_ref())
249            .push_bind(row.author_id().as_ref())
250            .push_bind(row.fhir_version())
251            .push_bind(FHIRJsonRef(row.resource()))
252            .push_bind(row.deleted())
253            .push_bind(row.request_method())
254            .push_bind(row.author_type().as_ref())
255            .push_bind(row.fhir_method());
256    });
257
258    query_builder
259        .build()
260        .execute(executor)
261        .await
262        .map_err(StoreError::from)?;
263
264    Ok(())
265}
haste_repository/pg/pending.rs

haste_repository/pg/
pending.rs
