#294 feat: domain operations on the transaction adapter

README.md

@@ -339,6 +339,18 @@ nothing else. The expressions land in the statement verbatim, like
 `#[column(default = "...")]`; the column names are checked against the
 entity at compile time.
 
+With `transactions` on, the same operation is available on the adapter,
+so it can land together with other writes — there it returns
+`Ok(None)` for a missing row, like the adapter's other methods:
+
+```rust,ignore
+let mut tx = pool.begin().await?;
+let verified = CitizenTransactionRepo::new(&mut tx)
+    .verify_passport(VerifyPassportCitizen { id, passport_provider: Some("gov".into()) })
+    .await?;
+tx.commit().await?;
+```
+
 ### Participant Scopes
 
 "Rows where this principal takes part in any role" is an OR over several

crates/entity-derive-impl/src/entity/transaction.rs

@@ -31,7 +31,7 @@
 use quote::{format_ident, quote};
 
 use super::{
-    parse::{EntityDef, SqlLevel, TransitionDef},
+    parse::{CommandSource, EntityDef, SqlLevel, TransitionDef},
     sql::postgres::Context
 };
 use crate::utils::{marker, tracing::instrument};
@@ -222,6 +222,7 @@
     };
 
     let transition_methods = transition_methods(entity, &ctx, error_type);
+    let domain_operation_methods = domain_operation_methods(entity, &ctx, error_type);
 
     let find_span = instrument(&entity_name_str, "tx.find_by_id");
     let find_for_update_span = instrument(&entity_name_str, "tx.find_by_id_for_update");
@@ -239,7 +240,8 @@
 
         /// Transaction repository adapter for #entity_name.
         ///
-        /// Provides repository operations that execute within an active transaction.
+        /// Provides repository operations that execute within an active transaction,
+        /// including the declared transitions and domain operations.
         /// Access via `ctx.{entities}()` within a transaction closure.
         ///
         /// Methods return the entity's configured `error` type; with
@@ -296,6 +298,8 @@
 
             #(#transition_methods)*
 
+            #(#domain_operation_methods)*
+
             /// Delete an entity within the transaction.
             #delete_span
             pub async fn delete(
@@ -323,6 +327,88 @@
     }
 }
 
+/// Generate the declared domain operations against the transaction.
+///
+/// Mirrors the pool repository's `#[command(..., sets(...))]` methods so
+/// an operation that must land together with other writes does not have
+/// to be rewritten as raw SQL. Same statement, same column checking; the
+/// only difference is the executor and that a missing row is `Ok(None)`
+/// rather than an error, matching the other adapter methods.
+fn domain_operation_methods(
+    entity: &EntityDef,
+    ctx: &Context<'_>,
+    error_type: &syn::Path
+) -> Vec<TokenStream> {
+    let entity_name = ctx.entity_name;
+    let entity_name_str = entity_name.to_string();
+    let row_name = &ctx.row_name;
+    let table = &ctx.table;
+    let id_name = &ctx.id_name;
+    let columns_str = &ctx.columns_str;
+    let soft_delete = ctx.soft_delete;
+
+    entity
+        .command_defs()
+        .iter()
+        .filter(|cmd| !cmd.sets.is_empty())
+        .map(|cmd| {
+            let method_name = format_ident!("{}", cmd.name.to_string().to_case(Case::Snake));
+            let command_struct = cmd.struct_name(&entity.name_str());
+            let payload: Vec<syn::Ident> = match &cmd.source {
+                CommandSource::Fields(fields) => fields.clone(),
+                _ => Vec::new()
+            };
+
+            let mut assignments: Vec<String> = cmd
+                .sets
+                .iter()
+                .map(|(column, expression)| format!("{column} = {expression}"))
+                .collect();
+            for (index, field) in payload.iter().enumerate() {
+                assignments.push(format!("{field} = ${}", index + 1));
+            }
+
+            let id_placeholder = payload.len() + 1;
+            let deleted_filter = if soft_delete {
+                " AND deleted_at IS NULL"
+            } else {
+                ""
+            };
+            let sql = format!(
+                "UPDATE {table} SET {} WHERE {id_name} = ${id_placeholder}{deleted_filter} \
+                 RETURNING {columns_str}",
+                assignments.join(", ")
+            );
+            let binds = payload
+                .iter()
+                .map(|field| quote! { .bind(&command.#field) });
+            let span = instrument(&entity_name_str, &format!("tx.{method_name}"));
+            let doc = format!(
+                "Run the `{}` operation within the transaction.\n\n\
+                 Writes the declared expressions plus the payload columns in\n\
+                 one UPDATE; returns `Ok(None)` when the row does not exist.",
+                cmd.name
+            );
+
+            quote! {
+                #[doc = #doc]
+                #span
+                pub async fn #method_name(
+                    &mut self,
+                    command: #command_struct,
+                ) -> Result<Option<#entity_name>, #error_type> {
+                    let row: Option<#row_name> = sqlx::query_as(#sql)
+                        #(#binds)*
+                        .bind(&command.id)
+                        .fetch_optional(&mut **self.tx)
+                        .await?;
+                    Ok(row.map(#entity_name::from))
+                }
+            }
+        })
+        .collect()
+}
+
 /// Generate `transition_to_{target}` methods from `#[transition(...)]`
 /// declarations.
 ///
@@ -715,6 +801,134 @@
     }
 }
 
+#[cfg(all(
+    test,
+    feature = "postgres",
+    feature = "transactions",
+    feature = "commands"
+))]
+mod tx_domain_operation_tests {
+    use quote::quote;
+    use syn::DeriveInput;
+
+    use super::*;
+
+    fn parse_entity(tokens: proc_macro2::TokenStream) -> EntityDef {
+        let input: DeriveInput = syn::parse2(tokens).expect("test entity must parse");
+        EntityDef::from_derive_input(&input).expect("test entity must be valid")
+    }
+
+    fn verified_citizen() -> EntityDef {
+        parse_entity(quote! {
+            #[entity(table = "citizens", transactions, commands)]
+            #[command(
+                VerifyPassport,
+                payload(passport_provider, passport_provider_id),
+                sets(passport_verified = "true", passport_verified_at = "NOW()")
+            )]
+            pub struct Citizen {
+                #[id]
+                pub id: uuid::Uuid,
+                #[field(create, update, response)]
+                pub name: String,
+                #[field(response)]
+                pub passport_verified: bool,
+                #[field(response)]
+                pub passport_provider: Option<String>,
+                #[field(response)]
+                pub passport_provider_id: Option<String>,
+                #[field(response)]
+                pub passport_verified_at: Option<chrono::DateTime<chrono::Utc>>,
+            }
+        })
+    }
+
+    #[test]
+    fn adapter_gains_the_declared_operation() {
+        let code = generate(&verified_citizen()).to_string();
+
+        assert!(code.contains("pub async fn verify_passport"));
+        assert!(code.contains("command : VerifyPassportCitizen"));
+        assert!(code.contains("self . tx"));
+    }
+
+    #[test]
+    fn the_statement_writes_expressions_then_payload_then_matches_the_id() {
+        let code = generate(&verified_citizen()).to_string();
+
+        assert!(code.contains(
+            "UPDATE citizens SET passport_verified = true, passport_verified_at = NOW(), \
+             passport_provider = $1, passport_provider_id = $2 WHERE id = $3"
+        ));
+    }
+
+    #[test]
+    fn every_payload_column_is_bound_in_order() {
+        let code = generate(&verified_citizen()).to_string();
+
+        let provider = code
+            .find("command . passport_provider)")
+            .expect("the first payload column must be bound");
+        let provider_id = code
+            .find("command . passport_provider_id)")
+            .expect("the second payload column must be bound");
+        assert!(
+            provider < provider_id,
+            "binds must follow the placeholder order"
+        );
+    }
+
+    #[test]
+    fn a_missing_row_is_not_an_error() {
+        let code = generate(&verified_citizen()).to_string();
+
+        assert!(code.contains("Result < Option < Citizen >"));
+        assert!(code.contains("fetch_optional"));
+    }
+
+    #[test]
+    fn a_command_without_sets_generates_no_operation() {
+        let entity = parse_entity(quote! {
+            #[entity(table = "citizens", transactions, commands)]
+            #[command(Deactivate, requires_id)]
+            pub struct Citizen {
+                #[id]
+                pub id: uuid::Uuid,
+                #[field(create, update, response)]
+                pub name: String,
+            }
+        });
+
+        let code = generate(&entity).to_string();
+
+        assert!(!code.contains("pub async fn deactivate"));
+    }
+
+    #[test]
+    fn soft_delete_keeps_deleted_rows_out_of_the_operation() {
+        let entity = parse_entity(quote! {
+            #[entity(table = "citizens", transactions, commands, soft_delete)]
+            #[command(VerifyPassport, payload(passport_provider), sets(passport_verified = "true"))]
+            pub struct Citizen {
+                #[id]
+                pub id: uuid::Uuid,
+                #[field(create, update, response)]
+                pub name: String,
+                #[field(response)]
+                pub passport_verified: bool,
+                #[field(response)]
+                pub passport_provider: Option<String>,
+                #[auto]
+                pub deleted_at: Option<chrono::DateTime<chrono::Utc>>,
+            }
+        });
+
+        let code = generate(&entity).to_string();
+
+        assert!(code.contains("WHERE id = $2 AND deleted_at IS NULL"));
+    }
+}
+
 #[cfg(all(test, feature = "postgres", feature = "transactions"))]
 mod tx_upsert_tests {
     use quote::quote;

crates/entity-derive/tests/postgres.rs

@@ -2826,7 +2826,7 @@ mod domain_operations {
     use crate::pg;
 
     #[derive(Debug, Clone, Entity)]
-    #[entity(table = "citizens", migrations, commands)]
+    #[entity(table = "citizens", migrations, commands, transactions)]
     #[command(
         VerifyPassport,
         payload(passport_provider),
@@ -2902,6 +2902,56 @@ mod domain_operations {
 
         db.teardown().await;
     }
+
+    #[tokio::test]
+    async fn the_operation_runs_inside_a_transaction() {
+        let Some(db) = pg::provision("domainoptx", &[Citizen::MIGRATION_UP]).await else {
+            return;
+        };
+        let pool = db.pool();
+
+        let citizen = pool
+            .create(CreateCitizenRequest {
+                name: "Grace".to_owned()
+            })
+            .await
+            .expect("create failed");
+
+        let mut tx = pool.begin().await.expect("begin failed");
+        let verified = CitizenTransactionRepo::new(&mut tx)
+            .verify_passport(VerifyPassportCitizen {
+                id:                citizen.id,
+                passport_provider: Some("gov".to_owned())
+            })
+            .await
+            .expect("the operation must apply")
+            .expect("the row exists");
+        assert!(verified.passport_verified);
+        tx.rollback().await.expect("rollback failed");
+
+        let after_rollback = pool
+            .find_by_id(citizen.id)
+            .await
+            .expect("lookup failed")
+            .expect("the row exists");
+        assert!(
+            !after_rollback.passport_verified,
+            "the operation must take part in the transaction, not commit on its own"
+        );
+
+        let mut tx = pool.begin().await.expect("begin failed");
+        let missing = CitizenTransactionRepo::new(&mut tx)
+            .verify_passport(VerifyPassportCitizen {
+                id:                Uuid::now_v7(),
+                passport_provider: None
+            })
+            .await
+            .expect("an unknown id is not an error here");
+        assert!(missing.is_none(), "an unknown id must report no row");
+        tx.commit().await.expect("commit failed");
+
+        db.teardown().await;
+    }
 }
 
 /// The hook-invoking wrapper: order of calls, and what a refusing

wiki/Comandos.md

@@ -156,6 +156,17 @@ Un comando con `sets(...)` escribe columnas nombradas directamente, sin que teng
 
 Un solo UPDATE escribe las expresiones fijas más las columnas del payload y nada más. Las expresiones llegan a la sentencia tal cual, como en `#[column(default = "...")]`; los nombres de columna se verifican contra la entidad en compilación.
 
+Con `transactions` activo, la misma operación está en el adaptador y se confirma junto con las demás escrituras; allí una fila ausente devuelve `Ok(None)`, como en los otros métodos del adaptador.
+
+```rust
+let mut tx = pool.begin().await?;
+let verified = UserTransactionRepo::new(&mut tx)
+    .verify_passport(VerifyPassportUser { id, passport_provider: Some("gov".into()) })
+    .await?;
+tx.commit().await?;
+```
+
+
 ### Comando Solo ID
 
 Añade solo el campo ID:

wiki/Commands-en.md

@@ -156,6 +156,17 @@ A command declaring `sets(...)` writes named columns directly, without them bein
 
 One UPDATE writes the fixed expressions plus the payload columns and nothing else. The expressions land in the statement verbatim, like `#[column(default = "...")]`; the column names are checked against the entity at compile time.
 
+With `transactions` on, the same operation is available on the adapter, so it can land together with other writes; there a missing row is `Ok(None)`, like the adapter's other methods.
+
+```rust
+let mut tx = pool.begin().await?;
+let verified = UserTransactionRepo::new(&mut tx)
+    .verify_passport(VerifyPassportUser { id, passport_provider: Some("gov".into()) })
+    .await?;
+tx.commit().await?;
+```
+
+
 ### ID-Only Command
 
 Adds only the ID field:

wiki/Команды.md

@@ -156,6 +156,17 @@ pub trait UserCommandHandler: Send + Sync {
 
 Один UPDATE пишет фиксированные выражения плюс колонки из payload и ничего больше. Выражения попадают в запрос дословно, как в `#[column(default = "...")]`; имена колонок проверяются по сущности на компиляции.
 
+С включёнными `transactions` та же операция доступна на адаптере — она попадает в общую транзакцию с остальными записями; отсутствующая строка там даёт `Ok(None)`, как и у прочих методов адаптера.
+
+```rust
+let mut tx = pool.begin().await?;
+let verified = UserTransactionRepo::new(&mut tx)
+    .verify_passport(VerifyPassportUser { id, passport_provider: Some("gov".into()) })
+    .await?;
+tx.commit().await?;
+```
+
+
 ### Команда только с ID
 
 Добавляет только поле ID: