cratestack_sqlx/query/write/
update.rs

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

cratestack_sqlx/query/write/
update.rs
