命令

快速开始

#[derive(Entity)]
#[entity(table = "users", commands)]
#[command(Register)]
#[command(UpdateEmail: email)]
#[command(Deactivate, requires_id)]
pub struct User {
 #[id]
 pub id: Uuid,
 #[field(create, update, response)]
 pub email: String,
 #[field(create, response)]
 pub name: String,
 #[field(response)]
 pub active: bool,
}

生成的代码

命令结构体

/// User上Register操作的命令载荷。
#[derive(Debug, Clone)]
pub struct RegisterUser {
 pub email: String,
 pub name: String,
}
/// User上UpdateEmail操作的命令载荷。
#[derive(Debug, Clone)]
pub struct UpdateEmailUser {
 pub id: Uuid,
 pub email: String,
}
/// User上Deactivate操作的命令载荷。
#[derive(Debug, Clone)]
pub struct DeactivateUser {
 pub id: Uuid,
}

命令枚举

/// User实体的命令枚举。
#[derive(Debug, Clone)]
pub enum UserCommand {
 Register(RegisterUser),
 UpdateEmail(UpdateEmailUser),
 Deactivate(DeactivateUser),
}
impl EntityCommand for UserCommand {
 fn kind(&self) -> CommandKind {
 match self {
 UserCommand::Register(_) => CommandKind::Create,
 UserCommand::UpdateEmail(_) => CommandKind::Update,
 UserCommand::Deactivate(_) => CommandKind::Custom,
 }
 }
 fn name(&self) -> &'static str {
 match self {
 UserCommand::Register(_) => "Register",
 UserCommand::UpdateEmail(_) => "UpdateEmail",
 UserCommand::Deactivate(_) => "Deactivate",
 }
 }
}

结果枚举

/// User命令执行的结果枚举。
#[derive(Debug, Clone)]
pub enum UserCommandResult {
 Register(User),
 UpdateEmail(User),
 Deactivate,
}

Handler Trait

/// 处理User命令的异步trait。
#[async_trait]
pub trait UserCommandHandler: Send + Sync {
 type Error: std::error::Error + Send + Sync;
 type Context: Send + Sync;
 /// 将命令分派到适当的处理器。
 async fn handle(&self, cmd: UserCommand, ctx: &Self::Context)
 -> Result<UserCommandResult, Self::Error>;
 /// 处理Register命令。
 async fn handle_register(&self, cmd: RegisterUser, ctx: &Self::Context)
 -> Result<User, Self::Error>;
 /// 处理UpdateEmail命令。
 async fn handle_update_email(&self, cmd: UpdateEmailUser, ctx: &Self::Context)
 -> Result<User, Self::Error>;
 /// 处理Deactivate命令。
 async fn handle_deactivate(&self, cmd: DeactivateUser, ctx: &Self::Context)
 -> Result<(), Self::Error>;
}

命令语法参考

基本命令

#[command(Register)]
// 生成:RegisterUser { email, name }

特定字段

#[command(UpdateEmail: email)]
// 生成:UpdateEmailUser { id, email }
#[command(UpdateProfile: name, bio, avatar)]
// 生成:UpdateProfileUser { id, name, bio, avatar }

领域操作

#[command(VerifyPassport, payload(passport_provider), sets(
 passport_verified = "true",
 passport_verified_at = "NOW()"
))]
// 生成:VerifyPassportUser { id, passport_provider }
// pool.verify_passport(command) -> User

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命令

#[command(Deactivate, requires_id)]
// 生成:DeactivateUser { id }
#[command(Delete, requires_id, kind = "delete")]
// 生成:DeleteUser { id },返回 ()

自定义载荷

pub struct TransferPayload {
 pub from_account: Uuid,
 pub to_account: Uuid,
 pub amount: i64,
}
#[command(Transfer, payload = "TransferPayload")]
// 直接使用TransferPayload

自定义结果

pub struct TransferResult {
 pub transaction_id: Uuid,
 pub success: bool,
}
#[command(Transfer, payload = "TransferPayload", result = "TransferResult")]
// 返回TransferResult而不是实体

Source选项

#[command(Create, source = "create")] // 使用#[field(create)]字段(默认)
#[command(Modify, source = "update")] // 使用#[field(update)]字段(可选)
#[command(Ping, source = "none")] // 无载荷字段

Kind提示

#[command(Create, kind = "create")] // 返回实体(默认)
#[command(Update, kind = "update")] // 返回实体
#[command(Remove, kind = "delete")] // 返回 ()
#[command(Process, kind = "custom")] // 从source推断

实现示例

use async_trait::async_trait;
struct UserHandler {
 pool: PgPool,
 email_service: EmailService,
}
struct RequestContext {
 user_id: Option<Uuid>,
 correlation_id: Uuid,
}
#[async_trait]
impl UserCommandHandler for UserHandler {
 type Error = AppError;
 type Context = RequestContext;
 async fn handle(&self, cmd: UserCommand, ctx: &Self::Context)
 -> Result<UserCommandResult, Self::Error>
 {
 match cmd {
 UserCommand::Register(c) => {
 let user = self.handle_register(c, ctx).await?;
 Ok(UserCommandResult::Register(user))
 }
 UserCommand::UpdateEmail(c) => {
 let user = self.handle_update_email(c, ctx).await?;
 Ok(UserCommandResult::UpdateEmail(user))
 }
 UserCommand::Deactivate(c) => {
 self.handle_deactivate(c, ctx).await?;
 Ok(UserCommandResult::Deactivate)
 }
 }
 }
 async fn handle_register(&self, cmd: RegisterUser, ctx: &Self::Context)
 -> Result<User, Self::Error>
 {
 // 验证
 if cmd.email.is_empty() {
 return Err(AppError::Validation("邮箱必填".into()));
 }
 // 创建用户
 let user = User {
 id: Uuid::now_v7(),
 email: cmd.email.to_lowercase(),
 name: cmd.name,
 active: true,
 };
 // 持久化
 sqlx::query(
 "INSERT INTO users (id, email, name, active) VALUES (1,ドル 2,ドル 3,ドル 4ドル)"
 )
 .bind(user.id)
 .bind(&user.email)
 .bind(&user.name)
 .bind(user.active)
 .execute(&self.pool)
 .await?;
 // 副作用
 self.email_service.send_welcome(&user.email).await?;
 Ok(user)
 }
 async fn handle_update_email(&self, cmd: UpdateEmailUser, ctx: &Self::Context)
 -> Result<User, Self::Error>
 {
 // 授权检查
 if ctx.user_id != Some(cmd.id) {
 return Err(AppError::Forbidden("无法更新其他用户的邮箱".into()));
 }
 // 更新
 let user: User = sqlx::query_as(
 "UPDATE users SET email = 1ドル WHERE id = 2ドル RETURNING *"
 )
 .bind(&cmd.email.to_lowercase())
 .bind(cmd.id)
 .fetch_one(&self.pool)
 .await?;
 // 发送验证
 self.email_service.send_verification(&user.email).await?;
 Ok(user)
 }
 async fn handle_deactivate(&self, cmd: DeactivateUser, ctx: &Self::Context)
 -> Result<(), Self::Error>
 {
 sqlx::query("UPDATE users SET active = false WHERE id = 1ドル")
 .bind(cmd.id)
 .execute(&self.pool)
 .await?;
 Ok(())
 }
}

使用命令

async fn register_user(
 handler: &impl UserCommandHandler,
 email: String,
 name: String,
) -> Result<User, AppError> {
 let cmd = RegisterUser { email, name };
 let ctx = RequestContext {
 user_id: None,
 correlation_id: Uuid::new_v4(),
 };
 match handler.handle(UserCommand::Register(cmd), &ctx).await? {
 UserCommandResult::Register(user) => Ok(user),
 _ => unreachable!(),
 }
}
// 或直接调用特定handler
async fn update_email(
 handler: &impl UserCommandHandler,
 user_id: Uuid,
 new_email: String,
 ctx: &RequestContext,
) -> Result<User, AppError> {
 let cmd = UpdateEmailUser {
 id: user_id,
 email: new_email,
 };
 handler.handle_update_email(cmd, ctx).await
}

EntityCommand Trait

use entity_derive::{EntityCommand, CommandKind};
let cmd = UserCommand::Register(register_data);
// 获取命令元数据
assert_eq!(cmd.name(), "Register");
assert!(matches!(cmd.kind(), CommandKind::Create));
// 模式匹配
match cmd.kind() {
 CommandKind::Create => println!("创建实体"),
 CommandKind::Update => println!("更新实体"),
 CommandKind::Delete => println!("删除实体"),
 CommandKind::Custom => println!("自定义操作"),
}

命令钩子

#[derive(Entity)]
#[entity(table = "orders", commands, hooks)]
#[command(Place)]
#[command(Cancel, requires_id)]
pub struct Order { /* ... */ }

#[async_trait]
pub trait OrderHooks: Send + Sync {
 type Error: std::error::Error + Send + Sync;
 // 标准CRUD钩子...
 // 命令特定钩子
 async fn before_command(&self, cmd: &OrderCommand) -> Result<(), Self::Error>;
 async fn after_command(&self, cmd: &OrderCommand, result: &OrderCommandResult) -> Result<(), Self::Error>;
}

最佳实践

CQRS模式

#[derive(Entity)]
#[entity(table = "orders", commands)]
#[projection(Summary: id, status, total_cents, created_at)]
#[projection(Details: id, status, items, shipping_address, total_cents)]
#[command(Place)]
#[command(Ship, requires_id)]
#[command(Cancel, requires_id)]
pub struct Order { /* ... */ }
// 命令(写入端)
let result = handler.handle(OrderCommand::Place(place_order), &ctx).await?;
// 查询(读取端)
let summary = repo.find_by_id_summary(order_id).await?;
let details = repo.find_by_id_details(order_id).await?;

另见