conti-backend/docs/11-cross-domain-collaboration.md

# 11. 跨域协作与聚合

## 决策

跨 domain 的**读**走契约模块（`-contract`）里的接口，跨 domain 的**写/状态联动**走 Spring `ApplicationEvent` + `@TransactionalEventListener`，暂不引入消息中间件。`bff-orchestration` / `workbench` 的并行聚合用一个专用的有界线程池，带上下文传播和整体超时预算，按 tile 局部降级。

这一篇填的是原来整套文档最大的一个空白：[04-security-auth.md](./04-security-auth.md) 里写了"切店 → 通知 webview-ticket 失效"走事件，但事件机制本身从来没有被定义过。

## 一、跨域读：契约模块

模块结构和约束见 [01-project-structure.md](./01-project-structure.md)，这里只讲使用规则。

```kotlin
// domains/identity-store-contract/src/main/kotlin/.../identitystore/contract/StoreQueryService.kt
interface StoreQueryService {
    fun listStoresByUserId(userId: Long): List<StoreInfo>
    fun findStore(storeId: Long): StoreInfo?
}

data class StoreInfo(
    val storeId: Long,
    val name: String,
    val code: String,
)
```

```kotlin
// domains/identity-store/src/main/kotlin/.../identitystore/application/StoreQueryServiceImpl.kt
@Service
class StoreQueryServiceImpl(private val storeRepository: StoreRepository) : StoreQueryService {
    override fun listStoresByUserId(userId: Long): List<StoreInfo> =
        storeRepository.findStoresByUserId(userId).map { StoreInfo(it.id, it.name, it.code) }
}
```

```kotlin
// domains/workbench/src/main/kotlin/.../workbench/application/WorkbenchAppService.kt
@Service
class WorkbenchAppService(
    private val storeQueryService: StoreQueryService,   // 注入的是契约里的接口，不是 identity-store 的实现类
) { ... }
```

### 契约设计的四条规则

1. **契约模型是独立的数据结构，不是 Entity 的别名**。`StoreInfo` 只含调用方真正需要的字段。让它跟着 `StoreEntity` 一起长，等于把内部表结构变成了对外承诺，改一个列名就要动三个 domain。
2. **契约只承诺调用方需要的最小能力**。不要一上来就写 `findAll()`、`update()`——契约里出现的每个方法都是未来的约束。
3. **契约变更要向后兼容**。加方法、加可空字段没问题；删方法、改语义要先确认所有调用方，跟对客户端的 API 一个待遇。
4. **契约模块不含实现、不依赖 Spring Web/JPA**（ArchUnit 会检查，见 [10-testing.md](./10-testing.md)）。

### 什么时候不该用契约，而应该重新划边界

如果 `workbench` 要用 `identity-store` 的契约方法超过五六个，甚至开始要求对方加"给我拼好这个结构"的定制方法，那说明**边界划错了**——这块逻辑本来就该在一边，或者本来就该单独成域。契约模块变厚是个明确的设计告警，不要靠往里加方法来消化它。

## 二、跨域写：领域事件

### 为什么写操作不走契约接口

`workbench` 直接调 `identityStore.doSomething()` 意味着：workbench 要知道 identity-store 内部该做什么，identity-store 的事务边界被外部方法调用拉长，而且以后每多一个关心"切店"的域，就要在切店逻辑里多加一行调用——切店代码变成一个不断膨胀的通知中心。

事件反转了这个依赖方向：**发布方不知道谁在听**。切店只管发一个"门店切换了"的事实，谁关心谁自己订阅。

### 事件定义放在契约模块

```kotlin
// domains/identity-store-contract/src/main/kotlin/.../identitystore/contract/StoreSwitchedEvent.kt
data class StoreSwitchedEvent(
    val userId: Long,
    val fromStoreId: Long?,
    val toStoreId: Long,
    val occurredAt: Instant,
)
```

事件类型必须放在契约模块，否则订阅方要 import 发布方的内部类型，边界又破了。

### 发布：在事务内发布

```kotlin
// domains/identity-store/src/main/kotlin/.../identitystore/application/StoreSwitchAppService.kt
@Service
class StoreSwitchAppService(
    private val events: ApplicationEventPublisher,
    private val clock: Clock,
) {
    @Transactional
    fun switchStore(userId: Long, targetStoreId: Long): StoreContext {
        val store = findAccessibleStore(userId, targetStoreId)
            ?: throw BusinessException(ErrorCode.STORE_NOT_ACCESSIBLE, "无权访问该门店", HttpStatus.FORBIDDEN)
        // ... 更新当前门店、重新签发 access token（见 04-security-auth.md）

        events.publishEvent(StoreSwitchedEvent(userId, currentStoreId, targetStoreId, clock.instant()))
        return context
    }
}
```

### 订阅：`@TransactionalEventListener`

```kotlin
// domains/webview-ticket/src/main/kotlin/.../webviewticket/application/StoreSwitchedListener.kt
@Component
class StoreSwitchedListener(private val ticketRepository: WebviewTicketRepository) {

    private val log = LoggerFactory.getLogger(javaClass)

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    fun onStoreSwitched(event: StoreSwitchedEvent) {
        runCatching { ticketRepository.revokeActiveTickets(event.userId) }
            .onFailure { log.error("切店后作废 webview 票据失败 userId={}", event.userId, it) }
    }
}
```

这段代码里每个注解和写法都在解决一个具体问题：

| 写法 | 解决什么 |
| --- | --- |
| `@TransactionalEventListener(AFTER_COMMIT)` | 普通 `@EventListener` 是**同步、在同一事务内**执行的。用它意味着"切店事务回滚了，但票据已经作废"这种不一致；`AFTER_COMMIT` 保证只在主事务真正提交后才触发 |
| `@Transactional(REQUIRES_NEW)` | `AFTER_COMMIT` 阶段原事务已经提交，此时**没有活跃事务**。不开新事务的话，监听器里的写操作要么报错要么自动提交，行为不可控 |
| `runCatching` + 记日志 | 监听器抛异常**不会**回滚主事务（主事务已提交），只会让这次副作用静默丢失。必须显式捕获并留下可排查的日志 |

### 同步还是异步

**默认同步**（`AFTER_COMMIT` 但仍在同一线程）。理由：同步下 traceId、`@RequestScope` 的门店上下文都还在，出问题能直接顺着日志查下去；异步则要额外处理上下文传播，而目前这些副作用（作废票据、清缓存）都很轻，没必要付这个复杂度。

只有当某个监听器确实耗时（比如要调外部系统）时才加 `@Async`，并且必须：指定专用线程池（不要用默认的 `SimpleAsyncTaskExecutor`，它每次新建线程且无上界）、加上下文传播的 `TaskDecorator`（见下文第三节）、想清楚失败后怎么办。

### 事件的可靠性边界（必须如实认识）

`ApplicationEvent` 是**进程内、内存中**的：应用在事件发出后、监听器执行前崩溃，这个事件就永久丢了，没有重试、没有补偿。

所以这套机制**只能用于"丢了不致命"的副作用**——作废一张票据（下次换票时本来也会重新校验）、清一个缓存、记一条审计日志。**不能用于**：扣款、发货、任何丢了会造成数据不一致且无法自愈的操作。

在当前这套系统里，跨域事件的用途就是切店后作废票据这一类，符合这个边界。

### 什么时候升级到消息中间件

出现下面任意一条，就该引入 RabbitMQ/Kafka + 事务性发件箱（transactional outbox），而不是继续给 `ApplicationEvent` 打补丁：

- 事件消费失败需要**自动重试**，或者需要死信队列；
- 事件丢失会造成**业务上的资金/库存不一致**；
- 消费方被拆成了**独立进程**（模块化单体拆微服务时的必然结果）；
- 需要一个事件被**多个消费组各自独立消费**，且各自有独立的消费进度。

事务性发件箱的做法（记在这里备查，现在不实施）：在业务事务里往 `outbox` 表插一条记录（与业务写在同一个事务，天然原子），另有一个轮询任务把 outbox 里的记录投递到消息中间件并标记已发送。它解决的是"写库成功但发消息失败"这个用 `AFTER_COMMIT` 无论如何都消除不掉的窗口。

现在不做的理由很直接：引入中间件意味着多一套需要部署、监控、排障的基础设施，而当前唯一的跨域事件场景丢了也不致命。**等到有第一个"丢了会出事"的事件时再做**，那时候需求也更清楚。

## 三、聚合：`bff-orchestration` 与 `workbench` 的并行 fan-out

首页要同时拉采购、保修、门店信息等多个 tile（架构图 Flow 3），串行调用意味着总耗时是各下游耗时之和。同步栈下必须显式用线程池做并行。

### 专用线程池

```kotlin
// domains/workbench/src/main/kotlin/.../workbench/infrastructure/config/WorkbenchExecutorConfig.kt
// 模块内的 @Configuration 一律放 infrastructure/config/（见 01-project-structure.md 的脚手架），
// 不要散在模块根包下——那样它既不属于任何一层，ArchUnit 的分层规则也管不到它。
@Configuration
class WorkbenchExecutorConfig {

    @Bean("workbenchExecutor")
    fun workbenchExecutor(): ThreadPoolTaskExecutor = ThreadPoolTaskExecutor().apply {
        corePoolSize = 8
        maxPoolSize = 16
        queueCapacity = 32                       // 有界！无界队列会让 maxPoolSize 永远不生效
        setThreadNamePrefix("workbench-")
        setRejectedExecutionHandler(ThreadPoolExecutor.CallerRunsPolicy())
        // 关键：把 MDC（traceId）和 RequestContext（门店上下文）带到子线程
        setTaskDecorator(ContextPropagatingTaskDecorator())
        setWaitForTasksToCompleteOnShutdown(true)
        setAwaitTerminationSeconds(20)           // 配合 09 的优雅停机
        initialize()
    }
}
```

四个必须这么写的点：

1. **必须是专用池，不能用公共 `@Async` 默认池**。聚合任务和别的异步任务共用一个池，一个下游变慢就会把池占满，波及所有异步任务——这正是舱壁模式要防的事（见 [05-integration-layer.md](./05-integration-layer.md)）。
2. **队列必须有界**。`ThreadPoolTaskExecutor` 的 `queueCapacity` 默认是 `Integer.MAX_VALUE`，即无界——后果是任务全部堆进队列，线程数永远不会从 core 涨到 max，然后在某次流量高峰把堆内存吃光。
3. **`CallerRunsPolicy`**：池满时任务退回调用线程（Tomcat 线程）自己执行。这是一种天然的背压——聚合变慢了，但不会丢请求、不会抛 `RejectedExecutionException`。
4. **`ContextPropagatingTaskDecorator`**：Micrometer 提供的上下文传播装饰器，把 MDC 里的 traceId、`@RequestScope` 的门店上下文搬到子线程。**没有它，子线程里的日志全部断链，且 `StoreContextHolder` 直接取不到值——这是同步栈下并行聚合最典型的翻车方式**（[08-observability.md](./08-observability.md) 附录也点了这一处）。

### 整体超时预算

```kotlin
// domains/workbench/src/main/kotlin/.../workbench/application/WorkbenchAppService.kt
@Service
class WorkbenchAppService(
    @Qualifier("workbenchExecutor") private val executor: Executor,
    private val f6ApiClient: F6ApiClient,
    private val o2oClient: O2OClient,
    private val storeQueryService: StoreQueryService,
) {
    private val log = LoggerFactory.getLogger(javaClass)

    companion object {
        private val TOTAL_BUDGET = Duration.ofSeconds(3)   // 整个首页接口的总预算
    }

    fun loadHomepage(userId: Long, storeId: Long): HomepageResponse {
        val deadline = System.nanoTime() + TOTAL_BUDGET.toNanos()

        val procurement = supply("procurement") { f6ApiClient.fetchProcurementList(storeId) }
        val warranty = supply("warranty") { o2oClient.fetchWarrantySummary(storeId) }
        val store = supply("store") { storeQueryService.findStore(storeId) }

        return HomepageResponse(
            procurement = await(procurement, deadline, "procurement"),
            warranty = await(warranty, deadline, "warranty"),
            store = await(store, deadline, "store"),
        )
    }

    private fun <T> supply(tile: String, block: () -> T): CompletableFuture<Tile<T>> =
        CompletableFuture.supplyAsync({
            runCatching(block)
                .map { Tile.ok(it) }
                .getOrElse { log.warn("tile={} 加载失败，降级", tile, it); Tile.degraded() }
        }, executor)

    private fun <T> await(future: CompletableFuture<Tile<T>>, deadlineNanos: Long, tile: String): Tile<T> {
        val remaining = deadlineNanos - System.nanoTime()
        if (remaining <= 0) return Tile.degraded()
        return runCatching { future.get(remaining, TimeUnit.NANOSECONDS) }
            .getOrElse { log.warn("tile={} 超出总预算，降级", tile); Tile.degraded() }
    }
}
```

**总预算不等于各下游超时之和**。三个下游各配 2 秒超时，串行最坏 6 秒、并行最坏 2 秒——但这算的是单次调用，叠上 `@Retry`（3 次）之后单个 tile 最坏可能到 6 秒。所以必须有一个独立于下游配置的**接口级总预算**（这里 3 秒），到点就把还没回来的 tile 全部降级返回。没有这道闸，首页接口的最坏耗时是由下游配置的乘积决定的，不可控。

### 局部降级的响应约定

降级必须**对客户端可见**，不能悄悄返回空数据——客户端要能区分"这块真的没数据"和"这块没拉到"，才能决定是显示空态还是显示"加载失败，点击重试"。

```kotlin
data class Tile<T>(
    val data: T?,
    val status: TileStatus,     // OK / DEGRADED
) {
    companion object {
        fun <T> ok(data: T) = Tile(data, TileStatus.OK)
        fun <T> degraded() = Tile<T>(null, TileStatus.DEGRADED)
    }
}
```

对应的响应体：

```json
{
  "code": 0,
  "data": {
    "procurement": { "status": "OK", "data": { "pendingCount": 12 } },
    "warranty": { "status": "DEGRADED", "data": null },
    "store": { "status": "OK", "data": { "storeId": 1, "name": "示例门店" } }
  },
  "traceId": "..."
}
```

规则：**只要主数据（门店上下文）拿到了，首页接口就返回 `code: 0`**，个别 tile 降级不会让整个接口失败。这正是架构图 Flow 3 里"局部降级"的含义——一个外部系统抖动不应该让用户连首页都打不开。

## 关键规则

- 跨 domain 读走 `-contract` 接口，跨 domain 写/状态联动走领域事件；两者都不允许直接 import 对方的内部类型。
- 事件类型定义在契约模块；监听器一律 `@TransactionalEventListener(AFTER_COMMIT)` + `@Transactional(REQUIRES_NEW)` + 自己兜住异常。
- `ApplicationEvent` 只用于"丢了不致命"的副作用；出现需要重试/不能丢的场景，升级到消息中间件 + 事务性发件箱，不要给现有机制打补丁。
- 并行聚合必须用专用**有界**线程池 + `ContextPropagatingTaskDecorator` + `CallerRunsPolicy`。
- 聚合接口必须有独立于下游配置的**总超时预算**，超时的 tile 降级返回而不是整体失败。
- 降级状态必须在响应里显式表达（`status: DEGRADED`），不能用空数据冒充。

## 待补充

- 各 tile 的具体超时预算分配（需要先有真实的下游耗时数据）。
- 首页聚合结果的本地缓存策略——见 [12-concurrency-and-scheduling.md](./12-concurrency-and-scheduling.md)。

## 参考链接

- [Spring: Application Events](https://docs.spring.io/spring-framework/reference/core/beans/context-introduction.html#context-functionality-events)
- [Spring: `@TransactionalEventListener`](https://docs.spring.io/spring-framework/reference/data-access/transaction/event.html)
- [Micrometer: Context Propagation](https://docs.micrometer.io/context-propagation/reference/)
- [microservices.io: Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html)
- [Simon Brown: Modular Monoliths](https://www.youtube.com/watch?v=5OjqD-ow8GE)