cratestack_sqlx/query/write/
update.rs

1//! Single-row UPDATE with optional version locking, policy, audit + events.
2
3use cratestack_core::{AuditOperation, CratestackContext, CratestackError, ModelEventKind};
4
5use crate::audit::{
6    RunInTxOutcome, build_audit_event, enqueue_audit_event, ensure_audit_table, fetch_before_image,
7};
8use crate::descriptor::ensure_event_outbox_table;
9use crate::{ModelDescriptor, SqlxRuntime, UpdateModelInput, sqlx};
10
11use super::preview::render_update_preview_sql;
12use super::update_exec::update_record_in_conn;
13
14#[derive(Debug, Clone)]
15pub struct UpdateRecord<'a, M: 'static, PK: 'static> {
16    pub(crate) runtime: &'a SqlxRuntime,
17    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
18    pub(crate) id: PK,
19}
20
21impl<'a, M: 'static, PK: 'static> UpdateRecord<'a, M, PK> {
22    pub fn set<I>(self, input: I) -> UpdateRecordSet<'a, M, PK, I> {
23        UpdateRecordSet {
24            runtime: self.runtime,
25            descriptor: self.descriptor,
26            id: self.id,
27            input,
28            if_match: None,
29        }
30    }
31}
32
33#[derive(Debug, Clone)]
34pub struct UpdateRecordSet<'a, M: 'static, PK: 'static, I> {
35    pub(crate) runtime: &'a SqlxRuntime,
36    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
37    pub(crate) id: PK,
38    pub(crate) input: I,
39    pub(crate) if_match: Option<i64>,
40}
41
42impl<'a, M: 'static, PK: 'static, I> UpdateRecordSet<'a, M, PK, I>
43where
44    I: UpdateModelInput<M>,
45{
46    /// Expected version for optimistic locking. Required on models
47    /// that declare `@version`; ignored otherwise.
48    pub fn if_match(mut self, expected: i64) -> Self {
49        self.if_match = Some(expected);
50        self
51    }
52
53    pub fn preview_sql(&self) -> String {
54        let values = self.input.sql_values();
55        let columns: Vec<&str> = values.iter().map(|v| v.column).collect();
56        render_update_preview_sql(
57            self.descriptor.table_name,
58            self.descriptor.primary_key,
59            self.descriptor.version_column,
60            &columns,
61            &self.descriptor.select_projection(),
62        )
63    }
64
65    /// Participates in a caller-supplied transaction. Neither the
66    /// `AuditSink` fan-out nor the event outbox drain happens here —
67    /// see `create.rs`'s `run_in_tx` doc comment for the full contract
68    /// and how a caller opts into both after their own commit.
69    pub async fn run_in_tx<'tx>(
70        self,
71        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
72        ctx: &CratestackContext,
73    ) -> Result<RunInTxOutcome<M>, CratestackError>
74    where
75        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
76        PK: Send + Clone + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
77    {
78        if self.descriptor.version_column.is_some() && self.if_match.is_none() {
79            return Err(CratestackError::PreconditionFailed(
80                "If-Match header required for versioned model".to_owned(),
81            ));
82        }
83        let emits_event = self.descriptor.emits(ModelEventKind::Updated);
84        let audit_enabled = self.descriptor.audit_enabled;
85        if emits_event {
86            ensure_event_outbox_table(&mut **tx).await?;
87        }
88        if audit_enabled {
89            ensure_audit_table(self.runtime, &mut **tx).await?;
90        }
91        let values = self.input.sql_values();
92        let keep = self
93            .runtime
94            .keeps_before(&mut **tx, self.descriptor, emits_event)
95            .await?;
96        let before_record = fetch_before_image(
97            &mut **tx,
98            self.descriptor,
99            self.id.clone(),
100            keep.is_some(),
101            &values,
102        )
103        .await?;
104        let record =
105            update_record_in_conn(tx, self.descriptor, self.id, self.input, ctx, self.if_match)
106                .await?;
107        if emits_event {
108            crate::descriptor::enqueue_model_update(
109                &mut **tx,
110                self.descriptor.event_target(),
111                &record,
112                before_record.as_ref(),
113                keep,
114            )
115            .await?;
116        }
117        let mut audit_event = None;
118        if audit_enabled {
119            let before_snapshot = before_record
120                .as_ref()
121                .and_then(|m| serde_json::to_value(m).ok());
122            let after = serde_json::to_value(&record).ok();
123            let event = build_audit_event(
124                self.descriptor,
125                AuditOperation::Update,
126                before_snapshot,
127                after,
128                ctx,
129            );
130            enqueue_audit_event(&mut **tx, &event).await?;
131            audit_event = Some(event);
132        }
133        Ok(RunInTxOutcome::new(
134            record,
135            audit_event.into_iter().collect(),
136        ))
137    }
138
139    pub async fn run(self, ctx: &CratestackContext) -> Result<M, CratestackError>
140    where
141        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
142        PK: Send + Clone + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
143    {
144        // Inside an `@isolation` procedure: run on its transaction and
145        // defer the post-commit fan-out (docs/design/procedure-isolation.md
146        // §4, §6).
147        if let Some(bound) = self.runtime.bound() {
148            let emits = self
149                .descriptor
150                .emits(cratestack_core::ModelEventKind::Updated);
151            let outcome = crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx))?;
152            return Ok(bound.settle(outcome, emits));
153        }
154        super::update_run::run_update(
155            self.runtime,
156            self.descriptor,
157            self.id,
158            self.input,
159            self.if_match,
160            ctx,
161        )
162        .await
163    }
164}
cratestack_sqlx/query/write/update.rs

cratestack_sqlx/query/write/
update.rs
