# Room Outbox Compensation Worker Development Guide

本文档定义 `room-service` 的 outbox 补偿 worker 如何接入进程生命周期、如何配置、如何重试，以及如何验证腾讯云 IM 同步失败后可由 `room_outbox` 补投。目标是收口 Phase 1 的房间系统消息可靠性，不扩展活动消费和 MQ 架构。

## Scope

本阶段只做：

| Item | Result |
| --- | --- |
| App 生命周期接入 | `room-service` 启动时启动 outbox worker，下线时先停止扫描新批次，再关闭 gRPC/MySQL/Redis |
| worker 配置 | 支持启停、扫描间隔、批量大小、单条投递超时和固定间隔重试策略 |
| 腾讯云 IM 补偿 | 同步 `PublishRoomEvent` 失败不回滚房间状态，pending outbox 后续补投到腾讯云 IM |
| 可验证测试 | 单测覆盖同步失败后状态已提交、outbox pending、worker 成功补投并标记 published |

本阶段不做：

| Not Included | Reason |
| --- | --- |
| activity-service 消费主循环 | 活动消费需要独立消费位点，不能复用 IM 补偿状态 |
| MQ/Redis Stream | 当前 MySQL outbox 已能满足 Phase 1 补偿 |
| 精确一次投递 | 腾讯云 IM 和进程崩溃窗口只能保证 at-least-once，客户端按 `event_id` 去重 |
| 多 sink 状态表 | 先把腾讯云 IM 补偿闭环跑通；activity/audit 后续新增自己的 consumption 表 |
| 指数退避和 poison 队列 | 当前 `room_outbox` 没有 `next_retry_at_ms` 或 poison 状态，V1 用固定间隔重试 |

## Current State

当前代码已有基础能力：

| Component | Current Fact |
| --- | --- |
| `room-service/internal/room/service/pipeline.go` | 命令成功后先写 command log/outbox，再尝试同步 `PublishRoomEvent`；同步成功 best-effort 标记 `published`，同步失败不回滚状态 |
| `room-service/internal/room/service/outbox_worker.go` | 已有 `RunOutboxWorker(ctx, options)` 持续循环和 `ProcessPendingOutbox(ctx, options)` 单批处理 |
| `room-service/internal/storage/mysql/repository.go` | 已有 `ListPendingOutbox`、`MarkOutboxPublished`、`MarkOutboxFailed` |
| `room-service/internal/integration/tencent_im.go` | 已能把 `EventEnvelope` 转换成腾讯云 IM 房间系统事件 |
| `room-service/internal/app/app.go` | 已在进程生命周期中启动 presence stale worker 和 outbox worker，下线时先停 worker 再关闭 gRPC/MySQL/Redis |

当前风险：

- 代码接入前已经残留的 `pending` 事件仍会被补投；V1 接受 at-least-once，客户端按 `event_id` 去重。
- `room_outbox.status` 目前只有 `pending/published`，适合先表达腾讯云 IM 补偿状态，不适合同时表达 activity/audit 消费状态。
- 多个 room-service 实例同时扫描同一批 pending 时可能重复投递；V1 接受 at-least-once，客户端按 `event_id` 去重。

## Target Semantics

房间命令成功后的语义：

```text
Room Cell command applied
  -> write command log
  -> write room_outbox status=pending
  -> persist snapshot if needed
  -> replace in-memory RoomState
  -> try synchronous PublishRoomEvent
  -> if sync publish success, best-effort mark that event published
  -> if sync publish fails, keep pending for worker
```

补偿 worker 语义：

```text
scan room_outbox where status=pending
  -> publish one EventEnvelope to configured OutboxPublisher
  -> success: status=published, last_error=NULL
  -> failure: status=pending, retry_count=retry_count+1, last_error=<short error>
  -> next scan retries failed pending records
```

重要边界：

- worker 不执行房间命令，不修改 Room Cell，不写 command log，不写 snapshot。
- worker 只处理已经持久化的 outbox envelope。
- worker 投递失败不能影响 `room-service` ready；腾讯云 IM 抖动时 room 状态服务仍应接收房间命令。
- `request_id` 仍然只做追踪；outbox 去重键是 `event_id`，房间命令幂等键是 `command_id`。
- `published` 在本阶段表示腾讯云 IM 补偿通道已处理，不表示 activity/audit 已消费。

## Configuration

在 `services/room-service/internal/config/config.go` 增加：

```go
type OutboxWorkerConfig struct {
    Enabled        bool          `yaml:"enabled"`
    PollInterval   time.Duration `yaml:"poll_interval"`
    BatchSize      int           `yaml:"batch_size"`
    PublishTimeout time.Duration `yaml:"publish_timeout"`
    RetryStrategy  string        `yaml:"retry_strategy"`
}
```

挂到 room-service 配置：

```go
type Config struct {
    // existing fields...
    OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"`
}
```

默认值：

| Field | Default | Meaning |
| --- | ---: | --- |
| `enabled` | `true` | 本地和生产默认启动补偿 worker；测试可显式关闭 |
| `poll_interval` | `1s` | 每轮扫描 pending outbox 的间隔 |
| `batch_size` | `100` | 每轮最多处理的 outbox 记录数 |
| `publish_timeout` | `3s` | 单条事件调用腾讯云 IM 的最大时间 |
| `retry_strategy` | `fixed_interval` | V1 只支持固定间隔重试，失败记录下一轮继续扫描 |

配置文件示例：

```yaml
outbox_worker:
  enabled: true
  poll_interval: "1s"
  batch_size: 100
  publish_timeout: "3s"
  retry_strategy: "fixed_interval"
```

配置规则：

- `poll_interval <= 0` 时使用默认值，不能导致 busy loop。
- `batch_size <= 0` 时使用默认值，不能全表扫描。
- `publish_timeout <= 0` 时使用默认值，不能让单条外部调用无限阻塞 worker。
- `retry_strategy` 只接受空值或 `fixed_interval`；未知值启动失败，避免配置写错后静默改变投递语义。
- `config.yaml`、`config.docker.yaml`、`config.tencent.example.yaml` 必须同步增加该配置。

## App Lifecycle

`room-service/internal/app.App` 需要新增 worker 生命周期字段：

```go
type App struct {
    // existing fields...
    workerCancel context.CancelFunc
    workerWG     sync.WaitGroup
}
```

启动顺序：

```text
1. 初始化 MySQL repository、Redis directory、wallet client、Tencent IM publisher
2. 创建 room service 和 gRPC server
3. Run 标记 gRPC serving
4. 创建 worker context
5. 启动 presence stale worker
6. 如果 outbox_worker.enabled=true，启动 outbox worker
7. 进入 grpcServer.Serve
```

关闭顺序：

```text
1. MarkDraining
2. cancel worker context，停止扫描新 outbox 批次
3. worker 当前正在处理的单条事件受 publish_timeout 控制，结束后退出
4. workerWG.Wait 等待后台 worker 停止
5. GracefulStop gRPC，等待已进入命令完成持久化
6. 关闭 wallet gRPC、MySQL、Redis
```

关闭规则：

- 不能在 worker 退出前关闭 MySQL，否则可能无法标记投递结果。
- worker 收到 cancel 后不再拉取新批次；已经进入投递的事件可以完成成功标记或失败标记。
- 如果进程被强杀，未标记成功的记录保持 `pending`，下次启动继续补偿。

## Worker API

在 `room/service` 增加持续循环方法：

```go
type OutboxWorkerOptions struct {
    PollInterval   time.Duration
    BatchSize      int
    PublishTimeout time.Duration
    RetryStrategy  string
}

func (s *Service) RunOutboxWorker(ctx context.Context, options OutboxWorkerOptions)
```

`RunOutboxWorker` 只负责循环、节流和退出：

```text
normalize options
process one batch immediately
start ticker
for each tick:
  if ctx done: return
  ProcessPendingOutbox(ctx, options)
```

`ProcessPendingOutbox` 建议改成 options 版本：

```go
func (s *Service) ProcessPendingOutbox(ctx context.Context, options OutboxWorkerOptions) error
```

处理规则：

- 每轮调用 `repository.ListPendingOutbox(ctx, batch_size)`。
- 每条记录单独创建 `context.WithTimeout(ctx, publish_timeout)`。
- `PublishOutboxEvent` 成功后调用 `MarkOutboxPublished`。
- `PublishOutboxEvent` 失败后调用 `MarkOutboxFailed`。
- 单条失败不阻断同批后续记录。
- `ListPendingOutbox` 或 `MarkOutboxPublished/Failed` 失败属于本轮 worker 错误，记录后等待下一轮。

## Sync Publish Interaction

为了避免 worker 对同步成功事件重复补投，命令链路需要在同步成功后 best-effort 标记 outbox：

```text
if result.applied && result.syncEvent != nil:
  err := syncPublisher.PublishRoomEvent(ctx, *result.syncEvent)
  if err == nil:
    repository.MarkOutboxPublished(ctx, result.syncEvent.EventID)
  if err != nil:
    keep room_outbox pending
```

约束：

- `MarkOutboxPublished` 失败不能让已经成功提交的房间命令返回失败；最多造成后续重复投递。
- 同步成功但进程在标记 published 前崩溃时，worker 后续可能重复投递；客户端必须用 `event_id` 去重。
- 对 `RoomCreated`、`RoomHeatChanged`、`RoomRankChanged` 这类不推 IM 的事件，`PublishOutboxEvent` 返回 no-op success 后可标记 `published`，避免长期 pending。

## Retry Strategy

V1 采用固定间隔无限重试：

| Case | Behavior |
| --- | --- |
| 腾讯云 IM 网络错误 | `retry_count + 1`，保留 `pending`，下一轮继续 |
| 腾讯云 IM 返回可重试错误 | `retry_count + 1`，保留 `pending`，下一轮继续 |
| envelope 反序列化失败 | worker 本轮返回错误，不删除记录；后续需要人工排查或补工具 |
| 不需要推 IM 的事件类型 | no-op success，标记 `published` |
| 进程退出 | 未成功标记的记录保持 `pending` |

为什么不加最大重试次数：

- 房间系统消息是状态变更的外部表现，不能因为达到次数就丢弃。
- 当前 schema 没有 poison 状态，贸然新增半成品状态会让恢复和运营查询复杂化。
- `retry_count` 和 `last_error` 已足够驱动告警；后续需要 poison 队列时再单独设计 `failed/next_retry_at_ms`。

## Observability

首版至少打结构化日志字段：

| Field | Meaning |
| --- | --- |
| `worker` | 固定为 `room_outbox` |
| `node_id` | 当前 room-service 节点 |
| `event_id` | outbox 事件 ID |
| `event_type` | RoomUserJoined、RoomGiftSent 等 |
| `room_id` | 事件所属房间 |
| `retry_count` | 当前失败次数 |
| `status` | pending 或 published |
| `error` | 最近一次失败原因，不能包含密钥或完整腾讯响应体 |

指标后续可补：

```text
room_outbox_pending_total
room_outbox_publish_success_total
room_outbox_publish_failure_total
room_outbox_publish_latency_ms
room_outbox_oldest_pending_age_ms
```

## Implementation Order

按以下顺序做，避免 worker 一启动就制造重复消息：

1. 扩展 room-service 配置结构和三份 YAML，增加 `outbox_worker` 配置。
2. 给 `room/service` 增加 `OutboxWorkerOptions` 和 `RunOutboxWorker`。
3. 调整 `ProcessPendingOutbox`，支持 `batch_size`、单条 `publish_timeout`、单条失败不中断批次。
4. 调整命令链路：同步 `PublishRoomEvent` 成功后 best-effort `MarkOutboxPublished(event_id)`；失败保持 pending。
5. 在 `app.Run` 启动 outbox worker，在 `app.Close` cancel 并等待 worker 退出。
6. 增加单测覆盖同步失败补偿、重试失败保留 pending、worker cancel 退出。
7. 更新 README 当前限制，说明 outbox worker 已接入生命周期。

## Acceptance Tests

必须覆盖：

| Test | Expected Result |
| --- | --- |
| 同步 IM 失败后命令成功 | `JoinRoom` 或 `SendGift` 返回成功，RoomSnapshot 已更新 |
| 同步 IM 失败后 outbox pending | 对应 `event_id` 仍为 `pending`，`retry_count` 未因同步失败被提前增加 |
| worker 补投成功 | fake outbox publisher 收到同一 `event_id`，repository 标记 `published` |
| worker 补投失败 | 记录仍为 `pending`，`retry_count+1`，`last_error` 有值 |
| 下一轮成功 | 第一次失败、第二次成功后状态变为 `published` |
| 同步 IM 成功 | 对应 `event_id` best-effort 标记 `published`，worker 不再扫描该事件 |
| worker cancel | cancel 后不再扫描新批次，当前投递受 `publish_timeout` 约束后退出 |
| 非 IM 事件 | `RoomCreated/RoomHeatChanged/RoomRankChanged` 不发送腾讯 IM，但会标记 `published` |

已有测试 `TestRoomOutboxCompensationWhenSyncPublishFails` 可以作为基础，但需要补齐持续 worker 和同步成功去重场景。

## Verification Commands

代码完成后至少运行：

```bash
go test ./services/room-service/internal/room/service
go test ./services/room-service/internal/config
go test ./services/room-service/internal/app
go test ./...
```

涉及配置和生命周期后运行：

```bash
docker compose config
docker compose build room-service
```

如果要验证本地启动链路：

```bash
docker compose up -d mysql redis wallet-service room-service
docker compose ps
docker compose down
```

## Done Definition

完成标准：

```text
room command committed
  -> outbox persisted
  -> sync Tencent IM publish fails
  -> HTTP/gRPC command response still succeeds
  -> outbox remains pending
  -> app-level worker scans pending
  -> worker publishes same event_id to Tencent IM publisher
  -> outbox status becomes published
  -> process shutdown stops worker before closing MySQL
```

该闭环稳定后，Phase 1 的“房间系统消息同步失败后可补投”才算完成。