代码结构：事务显式传到用例

更新于 · 在 sijie.xyz 查看原条目 ↗

状态： 已在 v0.1.76 发布（2026-09-27）—— 设计与落地记录见 StandMeet 仓库的 docs/design/event-bus-outbox-webhooks.md。

总线大部分就是"接口后面的 infra"（queue-behind-ports），没太多可说的；真正要定的是三件事：事务怎么到达用例、订阅方如何零胶水注册、副作用端口搬到哪里。

flowchart TB
  subgraph infra["internal/infra"]
    pg["pgstore<br/>InTx(ctx, Beginner, fn) · Nested · DBTX<br/>Listener · Notify"]
    ev["events<br/>Recorder · declarations · relay loop<br/>retention · reads · webhook signing"]
    jb["jobs<br/>Jobs · Inspector · Runtime · Kind · Periodic"]
    rv["jobs/river<br/>only importer of riverqueue"]
    se["sideeffect/mail · sideeffect/supplier"]
    dt["detach<br/>detach.Go"]
  end
  subgraph dom["internal/&lt;domain&gt;"]
    uc["usecase<br/>InTx → repo.With(tx) · Record · Enqueue"]
    repo["repo<br/>With(tx) · Nested for its own tx"]
    sub["subscriber<br/>subscriptions + job handlers"]
    fac["facade<br/>EventTypes · Subscriptions · JobKinds (data)"]
  end
  root["cmd/server/wire/periodic.go: one line per source, collects every declaration, hands EmbedAdmits to the webhook fan-out"]
  uc --> pg
  uc --> ev
  uc --> jb
  repo --> pg
  sub --> se
  sub --> repo
  sub --> uc
  ev --> pg
  jb --> rv
  ev --> jb
  fac --> sub
  root -.-> fac

事务怎么到达用例

原来事务是在 repo 内部开的，大多数写入是单条语句 autocommit，用例手里没有事务，所以显式的 Record 没法和业务写入进同一个事务。触发器捕获的行事件（two-sources-of-events）不受影响，语义事件受影响。

事务是显式参数，不进 ctx。 ctx 里已经有 14 处 WithValue；再放事务就是给它加一种隐式依赖——同一个 repo 调用进不进事务，取决于调用方 ctx 里恰好装了什么——这正是 ctx 变神类的路。

sequenceDiagram
  participant UC as usecase
  participant P as pgstore
  participant R as repo
  participant E as Recorder
  participant J as Jobs
  UC->>P: InTx(ctx, pool, fn)
  P->>UC: fn(tx), the transaction handed over as a parameter
  UC->>R: requests.With(tx).Create(ctx, …)
  UC->>E: events.With(tx).Record(ctx, ownerID, type, subject, data)
  UC->>J: jobs.With(tx).Enqueue(ctx, kind, args, opts)
  UC-->>P: fn returns nil
  P->>P: COMMIT (business rows + event + job land together)
  Note over UC,J: Who is in the transaction is visible at the call site, a call without With behaves as before
  • pgstore.InTx(ctx, db, fn)：在 db 上开事务，db 是一个 Beginner：连接池，或一个已打开的事务（那就是一个 savepoint）。fn 返回 nil 就提交，否则回滚；fn 里 panic 会回滚后再抛出。
  • pgstore.Nested(q, pool)：被 With(q) 绑定的 repo 需要自己开事务时用它：q 是事务就在里面开 savepoint，否则在连接池上开一个新事务。
  • With(tx)：每个 repo、Recorder、Jobs 都有，返回绑定到该事务的副本。原有方法签名不动；谁在事务里由编译器检查。
  • ctx 只装取消与截止时间：本计划不增加任何 ctx key。Record 要的 owner 等信息显式传参，不从 ctx 读。
  • 门禁 check-tx-only-via-pgstore.sh：.Begin(、BeginTx(、BeginFunc( 只能出现在 internal/infra/pgstore。现有的 11 处 Begin 都已改完（计划时估的是约 20 处）（no-bypass-by-structure）。
  • UT：fn 返回错误或 panic 时业务、事件、任务一起回滚；With(tx) 的副本不影响原对象；不经 With 的调用行为不变（events-test-plan）。

订阅方零胶水

各域把事件类型（event-model）、订阅和任务种类都声明成数据（和 dispatcher op 同样形状），经自己的 facade 暴露。组合根（cmd/server/wire/periodic.go）每个来源一行把它们收集起来，不关心任务具体做什么。唯一的专门接线是把 embed 范围函数（EmbedAdmits，来自 access 域）交给 owner 域的 webhook 扇出。

副作用端口搬了家

OutboundSender 原来在 cmd/server/port，后台 supplier 调用在 plugin/adapters。现在邮件端口是 internal/infra/sideeffect/mail；后台 supplier 调用是 internal/infra/sideeffect/supplier 里持久的 supplier.invoke 任务。错误表、lint 豁免、测试钩子都跟着搬了。只有 subscriber 包、infra 和 cmd/server 可以 import 它们。