summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--source/_posts/如何在-DDD-CQRS-中处理跨聚合的事务.md276
1 files changed, 276 insertions, 0 deletions
diff --git a/source/_posts/如何在-DDD-CQRS-中处理跨聚合的事务.md b/source/_posts/如何在-DDD-CQRS-中处理跨聚合的事务.md
new file mode 100644
index 00000000..34db4f7c
--- /dev/null
+++ b/source/_posts/如何在-DDD-CQRS-中处理跨聚合的事务.md
@@ -0,0 +1,276 @@
+---
+title: 如何在 DDD + CQRS 中处理跨聚合的事务
+date: 2026-05-28 17:30:57
+tags: [Technique]
+---
+
+假设系统里有一个模块，用来管理化合物从 hit、lead 到 candidate 的推进。
+
+药化团队合成了一批化合物。生物团队上传活性、选择性、ADME、早期毒理等实验结果。项目负责人看完某个化合物的数据包后，要做一个决策：
+
+- 推进为候选化合物。
+- 退回继续优化。
+- 暂停该方向。
+
+提交这个决策时，系统通常要做这些事：
+
+- 写入一条候选化合物推进决策。
+- 改变化合物当前研发阶段。
+- 如果推进成功，创建候选药物档案。
+- 关闭当前评审责任。
+- 锁定本次使用的实验数据包版本。
+- 写入审计事件，说明阶段变化依据哪一次数据评审。
+
+这些写入必须一起成功。不能出现决策已经提交，但化合物阶段没变, 或是候选药物档案已创建，但数据包没有锁定等问题。
+
+最直接的实现，是在 Command Handler 里注入 Prisma，然后开事务：
+
+```ts
+class SubmitCompoundPromotionDecisionHandler {
+ constructor(private readonly prisma: PrismaClient) {}
+
+ async execute(command: SubmitCompoundPromotionDecision) {
+ return this.prisma.$transaction(async (tx) => {
+ const decision = await tx.compoundDecision.create({
+ data: {
+ compoundId: command.compoundId,
+ result: command.result,
+ reason: command.reason,
+ decidedBy: command.actorUserId,
+ },
+ });
+
+ await tx.compound.update({
+ where: { id: command.compoundId },
+ data: { stage: command.result === "promote" ? "candidate" : "lead" },
+ });
+
+ if (command.result === "promote") {
+ await tx.developmentCandidate.create({
+ data: { compoundId: command.compoundId },
+ });
+ }
+
+ await tx.reviewAssignment.update(...);
+ await tx.auditEvent.create(...);
+
+ return decision;
+ });
+ }
+}
+```
+
+但它把几个层次混在了一起。
+
+Command Handler 本来应该编排业务流程，但现在它编排了数据库表的更新顺序。
+
+化合物阶段变化本来应该由 `Compound` 聚合判断。比如只有 `lead_optimization` 阶段的化合物才能推进到 `candidate`，已经暂停的化合物不能直接推进。现在这条规则可能散落在 `tx.compound.update` 前后的 if 判断里。
+
+事务边界本来属于用例。现在它由具体 ORM 的 `$transaction` 暴露在 Command Handler 里。换 ORM、拆 outbox、改审计事件写法，Command Handler 都要动。
+
+---
+
+如果把这段逻辑挪到 repository 里：
+
+```ts
+compoundRepository.submitPromotionDecision(command);
+```
+
+这看起来把 Prisma 从 Command Handler 里拿掉了。但如果 `compoundRepository` 内部仍然创建决策、更新化合物、创建候选药物档案、关闭责任、写审计事件，那只是把混乱换了一个地方，因为 Repository 的核心职责应该只是持久化聚合，例如：
+
+```ts
+compoundRepository.findById(id);
+compoundRepository.save(compound);
+compoundDecisionRepository.save(decision);
+developmentCandidateRepository.save(candidate);
+```
+
+如果一个 `CompoundRepository` 开始保存 `CompoundDecision`、`DevelopmentCandidate`、`ReviewAssignment`、`AuditEvent`，它就不再只是 Compound 的 repository。它变成了一个跨模块事务服务，只是名字还叫 repository。
+
+---
+
+所以这里我的解决方案之一是在 application 层定义一个 Unit of Work port，它表达的是一个具体业务动作：提交候选化合物推进决策。
+
+```ts
+export interface SubmitCompoundPromotionDecisionUnitOfWork {
+ submit(input: {
+ compoundId: string;
+ dataPackageId: string;
+ result: "promote" | "optimize" | "pause";
+ reason?: string;
+ actorUserId: string;
+ }): Promise<CompoundPromotionDecisionDto>;
+}
+```
+
+Command Handler 依赖这个端口，而不是依赖 Prisma。
+
+```ts
+class SubmitCompoundPromotionDecisionHandler {
+ constructor(
+ private readonly unitOfWork: SubmitCompoundPromotionDecisionUnitOfWork,
+ ) {}
+
+ async execute(command: SubmitCompoundPromotionDecision) {
+ return this.unitOfWork.submit({
+ compoundId: command.compoundId,
+ dataPackageId: command.dataPackageId,
+ result: command.result,
+ reason: command.reason,
+ actorUserId: command.actorUserId,
+ });
+ }
+}
+```
+
+这段代码没有事务细节。它只转交命令意图。
+
+具体事务放在 infrastructure：
+
+```ts
+class PrismaSubmitCompoundPromotionDecisionUnitOfWork
+ implements SubmitCompoundPromotionDecisionUnitOfWork
+{
+ constructor(private readonly prisma: PrismaClient) {}
+
+ async submit(input: SubmitCompoundPromotionDecisionInput) {
+ return this.prisma.$transaction(async (tx) => {
+ // 在这里组合 tx-bound store / repository
+ });
+ }
+}
+```
+
+这样就把“业务事务边界”和“命令入口”分开，Command Handler 不需要知道用 Prisma 还是别的 ORM。它也不需要知道事务里要调用哪些表。它只知道这个命令由一个原子用例完成。
+
+但聚合仍然要负责状态规则，Unit of Work 不能变成新的数据库脚本，在事务内部，仍然应该先读取聚合，让聚合执行状态变化，再保存聚合。
+
+例如 `Compound` 聚合可以这样表达规则：
+
+```ts
+class Compound {
+ promoteToCandidate(input: {
+ dataPackageId: string;
+ actorUserId: string;
+ }) {
+ if (this.stage !== "lead_optimization") {
+ throw new DomainError("只有先导优化阶段的化合物才能推进为候选化合物");
+ }
+
+ if (this.lockedDataPackageId !== null) {
+ throw new DomainError("已锁定数据包的化合物不能重复推进");
+ }
+
+ this.stage = "candidate";
+ this.lockedDataPackageId = input.dataPackageId;
+ this.updatedByUserId = input.actorUserId;
+ }
+
+ returnToOptimization(input: { reason: string; actorUserId: string }) {
+ if (this.stage !== "lead_optimization") {
+ throw new DomainError("只有评审中的化合物才能退回优化");
+ }
+
+ this.stage = "lead_optimization";
+ this.optimizationReason = input.reason;
+ this.updatedByUserId = input.actorUserId;
+ }
+}
+```
+
+Unit of Work 在事务里调用这些方法：
+
+```ts
+await prisma.$transaction(async (tx) => {
+ const compoundStore = new PrismaCompoundStore(tx);
+ const compound = await compoundStore.findById(input.compoundId);
+
+ if (input.result === "promote") {
+ compound.promoteToCandidate({
+ dataPackageId: input.dataPackageId,
+ actorUserId: input.actorUserId,
+ });
+ }
+
+ if (input.result === "optimize") {
+ compound.returnToOptimization({
+ reason: input.reason ?? "需要补充结构优化",
+ actorUserId: input.actorUserId,
+ });
+ }
+
+ await compoundStore.save(compound);
+ await decisionStore.save(decision);
+ await candidateStore.createIfPromoted(...);
+ await assignmentStore.closeCurrent(...);
+ await auditStore.append(...);
+});
+```
+
+而为了让事务里的代码复用聚合持久化逻辑，我们需要一种 tx-bound repository 或 store。
+
+最粗暴的写法是在领域 repository interface 里加一个可选事务参数：
+
+```ts
+interface CompoundRepository {
+ findById(id: string, tx?: Prisma.TransactionClient): Promise<Compound | null>;
+ save(compound: Compound, tx?: Prisma.TransactionClient): Promise<void>;
+}
+```
+
+但它又把 Prisma 带进了领域层。就算把 `Prisma.TransactionClient` 包成 `TransactionClient` 类型别名，领域接口仍然在为基础设施妥协。
+
+我更倾向于把 tx-bound 能力留在 infrastructure，领域层只定义干净的 repository port：
+
+```ts
+export interface CompoundRepository {
+ findById(id: string): Promise<Compound | null>;
+ save(compound: Compound): Promise<void>;
+}
+```
+
+infrastructure 层拆一个 store，接收最小持久化客户端：
+
+```ts
+type CompoundStoreClient = {
+ compound: {
+ findUnique(args: unknown): Promise<unknown>;
+ update(args: unknown): Promise<unknown>;
+ };
+};
+
+class PrismaCompoundStore implements CompoundRepository {
+ constructor(private readonly client: CompoundStoreClient) {}
+
+ async findById(id: string) {
+ const record = await this.client.compound.findUnique(...);
+ return Compound.restore(record);
+ }
+
+ async save(compound: Compound) {
+ const snapshot = compound.toSnapshot();
+ await this.client.compound.update(...);
+ }
+}
+```
+
+普通 repository 可以继承这个 store，传完整 Prisma client：
+
+```ts
+class PrismaCompoundRepository extends PrismaCompoundStore {
+ constructor(prisma: PrismaClient) {
+ super(prisma);
+ }
+}
+```
+
+Unit of Work 在事务里传 tx：
+
+```ts
+await prisma.$transaction(async (tx) => {
+ const compoundStore = new PrismaCompoundStore(tx);
+ const compound = await compoundStore.findById(input.compoundId);
+ // ...
+ await compoundStore.save(compound);
+});
+```
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-04 08:39:37 +0000