cratestack_sqlx/query/write/
update_exec.rs

1//! Generic-over-Executor update helpers used by single-row UPDATE
2//! paths. Builds `UPDATE ... SET ... WHERE pk = $X [AND version = $Y]
3//! AND policy(...) RETURNING ...`, with version-mismatch detection via
4//! a read-policy probe.
5
6use cratestack_core::{CratestackContext, CratestackError};
7
8use crate::query::support::{
9    classify_unique_violation, probe_current_version, push_action_policy_query, push_bind_value,
10};
11use crate::{ModelDescriptor, UpdateModelInput, sqlx};
12
13pub async fn update_record_with_executor<'e, E, M, PK, I>(
14    executor: E,
15    policy_pool: &sqlx::PgPool,
16    descriptor: &'static ModelDescriptor<M, PK>,
17    id: PK,
18    input: I,
19    ctx: &CratestackContext,
20    if_match: Option<i64>,
21) -> Result<M, CratestackError>
22where
23    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
24    I: UpdateModelInput<M>,
25    for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
26    PK: Send + Clone + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
27{
28    input.validate()?;
29    let values = input.sql_values();
30    if values.is_empty() {
31        return Err(CratestackError::Validation(
32            "update input must contain at least one changed column".to_owned(),
33        ));
34    }
35
36    update_returning_record(
37        executor,
38        policy_pool,
39        descriptor,
40        id,
41        &values,
42        ctx,
43        if_match,
44    )
45    .await
46}
47
48#[allow(clippy::too_many_arguments)]
49async fn update_returning_record<'e, E, M, PK>(
50    executor: E,
51    policy_pool: &sqlx::PgPool,
52    descriptor: &'static ModelDescriptor<M, PK>,
53    id: PK,
54    values: &[crate::SqlColumnValue],
55    ctx: &CratestackContext,
56    if_match: Option<i64>,
57) -> Result<M, CratestackError>
58where
59    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
60    for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow>,
61    PK: Send + Clone + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
62{
63    let version_column = descriptor.version_column;
64    let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("UPDATE ");
65    query.push(descriptor.table_name).push(" SET ");
66    for (index, value) in values.iter().enumerate() {
67        if index > 0 {
68            query.push(", ");
69        }
70        query.push(value.column).push(" = ");
71        push_bind_value(&mut query, &value.value);
72    }
73    if let Some(version_col) = version_column {
74        query
75            .push(", ")
76            .push(version_col)
77            .push(" = ")
78            .push(version_col)
79            .push(" + 1");
80    }
81    query
82        .push(" WHERE ")
83        .push(descriptor.primary_key)
84        .push(" = ");
85    let id_for_probe = id.clone();
86    query.push_bind(id);
87    if let (Some(version_col), Some(expected)) = (version_column, if_match) {
88        query.push(" AND ").push(version_col).push(" = ");
89        query.push_bind(expected);
90    }
91    query.push(" AND ");
92    push_action_policy_query(
93        &mut query,
94        descriptor.update_allow_policies,
95        descriptor.update_deny_policies,
96        ctx,
97    );
98    query
99        .push(" RETURNING ")
100        .push(descriptor.select_projection());
101
102    let outcome = query
103        .build_query_as::<M>()
104        .fetch_optional(executor)
105        .await
106        .map_err(classify_unique_violation)?;
107    match outcome {
108        Some(record) => Ok(record),
109        None => {
110            // If this is a versioned update, distinguish "stale
111            // version" from a true policy denial via the read-policy
112            // probe. If the caller can't see the row, we keep
113            // returning Forbidden so policy denials remain
114            // indistinguishable from missing rows.
115            if let (Some(version_col), Some(expected)) = (version_column, if_match)
116                && let Some(current) =
117                    probe_current_version(policy_pool, descriptor, id_for_probe, version_col, ctx)
118                        .await?
119                && current != expected
120            {
121                return Err(CratestackError::PreconditionFailed(format!(
122                    "version mismatch: expected {expected}, found {current}",
123                )));
124            }
125            Err(CratestackError::Forbidden(
126                "update policy denied this operation".to_owned(),
127            ))
128        }
129    }
130}
cratestack_sqlx/query/write/update_exec.rs

cratestack_sqlx/query/write/
update_exec.rs
