CommandService

包：command

command.Service 是写操作的唯一入口。一次 Execute 就是一个数据库事务加一个 UnitOfWork。

结构

type Service struct {
    Middleware []Middleware
    Filters    []Filter
    Router     *Router
    Repo       Committer         // 通常是 *aggregate.Repository
    Tx         eventstore.TxManager
    Obs        *obs.Instruments  // 可选
}

err := svc.Execute(ctx, cmd1, cmd2) // 一个事务，一个 UoW

Router 与 Tx 必填，任一为 nil 时 Execute 返回错误。没有命令时直接返回 nil。

通常用 sequent.New 拿到配好的 service，见 接线。

Execute 流程

1. 把命令复制进队列。
2. 在 context 上创建 AfterCommit 钩子容器。
3. 设置了 Obs 时，开始 span 与计时。
4. Tx.WithinTx 开启事务。
5. 创建 UnitOfWork 并挂到 context。
6. 队列非空时，取出下一条命令，执行下面的步骤。
7. 队列清空后，事务提交。
8. 记录遥测数据。
9. 没有错误时，执行 AfterCommit 钩子。

每条命令：

EnsureMeta
  → 中间件链（外层先行）
    → Filters（AuthZ）
      → Validate（命令实现了才执行）
        → Router.Dispatch → Handler
          → Repo.Commit

任何错误都会中止整次调用：事务回滚，什么都不落库，AfterCommit 钩子也不会跑。

中间件

type Middleware func(ctx context.Context, cmd Command, next HandlerFunc) error

Chain 外层先行：下标 0 最先执行。典型顺序：

1. tenancy.Middleware：把 context 中的租户抄到 Meta.TenantID。
2. Idempotency 中间件：检查 (TenantID, IdempotencyKey)（同一业务键重复提交只生效一次）。

tenancy 要放在 Idempotency 前面，因为幂等键空间依赖租户。bank 演示就是这样做的。sequent.WithIdempotency 会追加到 Service.Middleware，所以需要自己把 tenancy 放到最前面：

app.Service.Middleware = append(
    []command.Middleware{tenancy.Middleware},
    app.Service.Middleware...,
)

中间件可以不调用 next 直接返回来短路。Idempotency 中间件对重放的键就是这样做的。

过滤器

type Filter interface {
    Filter(ctx context.Context, cmd Command) error
}

过滤器在中间件链内部、校验之前运行。sequent.WithAuthorizer(auth) 会安装 security.Filter{Auth: auth}，拒绝时返回 security.ErrDenied。见 AuthZ。

校验与分发

命令实现了 Validator 时执行 Validate()。随后 Router.Dispatch 按 CommandType() 查找 Handler，找不到就是错误。

提交

分发之后，Repo.Commit(ctx, RecordOf(cmd)) 追加事件并发布。如果错误是 ErrOptimisticConcurrency 且设置了 Obs，service 会先调用 RecordConflict。见 Repository。

命令队列

Workflow 可以在事务内再追加一条命令：

command.Enqueue(ctx, followUp)

入队的命令排在队列已有命令之后，在同一事务与 UnitOfWork 中执行。见 Workflow。

AfterCommit 钩子

用 workflow.RegisterAfterCommit(ctx, fn) 注册的钩子只在事务提交后执行。钩子的错误会从 Execute 返回，但已提交的事件保持不变。不应回滚账本的工作适合放在钩子里。

批量执行

svc.Execute(ctx, cmd1, cmd2, cmd3)

所有命令共用一个事务。批量调用的 OpenTelemetry 标签是第一条命令的类型加 +batch。任何一条失败，全部不落库。

可观测性

Obs 是一个字段，不是中间件。用 sequent.WithObs(inst) 设置，或直接赋值 svc.Obs = inst。见 OpenTelemetry。

错误处理

err := svc.Execute(ctx, cmd)
switch {
case errors.Is(err, eventstore.ErrOptimisticConcurrency):
    // 带退避重试，重新 Load
case errors.Is(err, eventstore.ErrUniqueKeyViolation):
    // 业务重复
case errors.Is(err, security.ErrDenied):
    // 授权失败
case idempotency.IsReplay(err):
    // Strict 模式：键已处理过
}

相关

• 核心闭环
• CommandHandler
• 写路径一致性

文档随 go-sequent 源码更新；若与 API 不一致，以源码为准。