Workflow

包：workflow

Workflow 响应事件来编排后续工作：可以追加命令、记录状态，并注册只在事务提交之后才执行的副作用。

接口

Workflow 就是一个 publish.Handler：

type Workflow interface {
    publish.Handler // Handles() []string；Handle(ctx, event.Event) error
}

通过门面注册：

app.RegisterWorkflow(&NotifyWorkflow{})

Workflow 在 SyncInTxn（在写事务里同步更新读模型）上运行，晚于同一事件的 Projector（把事件写成查询用读模型的组件）。它们不参与读模型重建：replay 流水线使用 PublishProjectorsOnly，所以重建不会重复发通知，也不会重复发命令。见 Publisher。

AfterCommit

无法回滚的工作——邮件、webhook、调用支付服务商——不能放在事务里跑。如果它已经执行而事务随后回滚，你就把一件没发生的事告诉了外部世界。

请改为注册 AfterCommit 钩子：

workflow.RegisterAfterCommit(ctx, func(ctx context.Context) error {
    // 副作用
    return nil
})

钩子按每次 Execute 收集在 context 上，按注册顺序执行，且只在事务成功提交之后。Execute 因任何原因失败时，一个钩子都不会跑。

钩子失败时，错误会从 Execute 返回，但已提交的事件保持提交状态。钩子失败不会撤销账本写入。请把钩子当作尽力而为；需要重试时用 Outbox。

Bank 示例中的写法

type NotifyWorkflow struct {
    mu       sync.Mutex
    Opened   []string
    Notified []string
}

func (w *NotifyWorkflow) Handles() []string { return []string{EventAccountOpened} }

func (w *NotifyWorkflow) Handle(ctx context.Context, e event.Event) error {
    ev := e.(*AccountOpened)
    w.mu.Lock()
    w.Opened = append(w.Opened, ev.Owner)
    owner := ev.Owner
    w.mu.Unlock()

    workflow.RegisterAfterCommit(ctx, func(ctx context.Context) error {
        w.mu.Lock()
        w.Notified = append(w.Notified, owner)
        w.mu.Unlock()
        return nil
    })
    return nil
}

AccountOpened 发布时，Workflow 立刻记下 owner，并把通知安排到提交之后。测试可以据此断言：只有命令成功时 Notified 才会有内容。

追加命令

Workflow 可以在同一事务里发起更多工作：

command.Enqueue(ctx, &SomeFollowUp{ /* ... */ })

该命令排在队列已有命令之后，在同一个 UnitOfWork 与事务中运行。请把 Meta.Source 设为 workflow 并填好因果字段，这样审计记录能说明命令为何运行。

Workflow 还是 Outbox

如果 Workflow 必须调用 Circle API 这类外部服务，请放在 AfterCommit 钩子或 Outbox 投递里，让外部延迟和重试远离账本事务。

规则

• Workflow 在写事务内运行。只有想回滚账本写入时才返回错误。
• 重建时不会重放 Workflow，所以不要在里面维护读模型；用 Projector。
• 尽量让 Workflow 处理器幂等（重复执行也不会产生额外副作用）。命令因冲突重试时它们可能再次运行。

相关

• Publisher
• CommandService
• 读模型重建

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