cratestack_sqlx/query/write/
create_exec.rs

1//! Generic-over-Executor create helper used by both the pool and
2//! transaction paths in [`super::create`]. Validates, applies
3//! auth-defaults, seeds `@version`, evaluates create policies, then
4//! runs `INSERT ... RETURNING`.
5
6use cratestack_core::{CratestackContext, CratestackError};
7
8use crate::query::support::{
9    PolicyDb, apply_create_defaults, classify_unique_violation, evaluate_create_policies,
10    find_column_value, push_bind_value,
11};
12use crate::{CreateModelInput, ModelDescriptor, sqlx};
13
14/// Passing a pool as `policy_pool` while `executor` is a transaction takes a
15/// second connection; prefer the builder's `run_in_tx`.
16pub async fn create_record_with_executor<'e, E, M, PK, I>(
17    executor: E,
18    policy_pool: &sqlx::PgPool,
19    descriptor: &'static ModelDescriptor<M, PK>,
20    input: I,
21    ctx: &CratestackContext,
22) -> Result<M, CratestackError>
23where
24    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
25    I: CreateModelInput<M>,
26    for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
27{
28    let values =
29        authorized_create_values(PolicyDb::Pool(policy_pool), descriptor, input, ctx).await?;
30    insert_returning_record(executor, descriptor, &values).await
31}
32
33/// [`create_record_with_executor`] for a write that runs on `conn`, with
34/// the create-policy evaluation on `conn` too
35/// (docs/design/procedure-isolation.md §4.1).
36pub(crate) async fn create_record_in_conn<M, PK, I>(
37    conn: &mut sqlx::PgConnection,
38    descriptor: &'static ModelDescriptor<M, PK>,
39    input: I,
40    ctx: &CratestackContext,
41) -> Result<M, CratestackError>
42where
43    I: CreateModelInput<M>,
44    for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
45{
46    let values =
47        authorized_create_values(PolicyDb::Conn(&mut *conn), descriptor, input, ctx).await?;
48    insert_returning_record(&mut *conn, descriptor, &values).await
49}
50
51async fn authorized_create_values<M, PK, I>(
52    policy: PolicyDb<'_>,
53    descriptor: &'static ModelDescriptor<M, PK>,
54    input: I,
55    ctx: &CratestackContext,
56) -> Result<Vec<crate::SqlColumnValue>, CratestackError>
57where
58    I: CreateModelInput<M>,
59{
60    input.validate()?;
61    let mut values = apply_create_defaults(input.sql_values(), descriptor.create_defaults, ctx)?;
62    descriptor.fill_generated_defaults(&mut values);
63    // Seed the optimistic-lock column server-side. `@version` is
64    // excluded from the generated Create input so clients can't pick
65    // the initial value, and the column has no SQL `DEFAULT`. Done
66    // after `apply_create_defaults` so `@default`-driven overrides
67    // still win if a schema ever lands one.
68    if let Some(version_col) = descriptor.version_column
69        && find_column_value(&values, version_col).is_none()
70    {
71        values.push(crate::SqlColumnValue {
72            column: version_col,
73            value: crate::SqlValue::Int(0),
74        });
75    }
76    if values.is_empty() {
77        return Err(CratestackError::Validation(
78            "create input must contain at least one column".to_owned(),
79        ));
80    }
81    if !evaluate_create_policies(
82        policy,
83        descriptor.create_allow_policies,
84        descriptor.create_deny_policies,
85        &values,
86        ctx,
87    )
88    .await?
89    {
90        return Err(CratestackError::Forbidden(
91            "create policy denied this operation".to_owned(),
92        ));
93    }
94
95    Ok(values)
96}
97
98async fn insert_returning_record<'e, E, M, PK>(
99    executor: E,
100    descriptor: &'static ModelDescriptor<M, PK>,
101    values: &[crate::SqlColumnValue],
102) -> Result<M, CratestackError>
103where
104    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
105    for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow>,
106{
107    let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("INSERT INTO ");
108    query.push(descriptor.table_name).push(" (");
109    for (index, value) in values.iter().enumerate() {
110        if index > 0 {
111            query.push(", ");
112        }
113        query.push(value.column);
114    }
115    query.push(") VALUES (");
116    for (index, value) in values.iter().enumerate() {
117        if index > 0 {
118            query.push(", ");
119        }
120        push_bind_value(&mut query, &value.value);
121    }
122    query
123        .push(") RETURNING ")
124        .push(descriptor.select_projection());
125
126    query
127        .build_query_as::<M>()
128        .fetch_one(executor)
129        .await
130        .map_err(classify_unique_violation)
131}
cratestack_sqlx/query/write/create_exec.rs

cratestack_sqlx/query/write/
create_exec.rs
