# Integration Inbox/Outbox 架构重构方案

> 状态: **Implemented** | 日期: 2026-04-11 | 实施: 2026-04-11 | Commit: `7bfc2e3f`

## 1. 问题陈述

当前 `_System_OutboxEvent` 单表承担三种职责：

| 职责 | EventType 示例 | 消费者 |
|------|---------------|--------|
| 内部 CRM 事件 | `record.created`, `record.updated` | ProcessOutbox → EventBus |
| 集成入站 (外部→CRM) | `integration.hubspot.contact.update` | SyncWorker |
| 集成出站 (CRM→外部) | _(未实现，OutboundWorker 用 EventBus 内存模式)_ | OutboundWorker |

### 1.1 现有问题

| # | 问题 | 代码位置 | 影响 |
|---|------|---------|------|
| P1 | **前缀路由 hack** — ProcessOutbox 用 `strings.HasPrefix("integration.")` 跳过集成事件 | `outbox_service.go:139` | 隐式契约，新 connector 必须遵守前缀约定 |
| P2 | **LIKE 查询不走索引** — SyncWorker 用 `EventType LIKE 'integration.hubspot.%'` 拉取事件 | `outbox_repository.go:171-179` | 表增大后全表扫描 |
| P3 | **OutboundWorker 内存累积器数据丢失** — 订阅 EventBus，5 秒窗口内 crash 丢失所有未刷出变更 | `outbound_worker.go:180,186-224` | 生产数据丢失风险 |
| P4 | **入站事件缺少一等列** — CanonicalEvent 的 ExternalID/Provider/RawPayload 序列化进 JSON blob，无法索引 | `canonical_event.go:10-32` | 去重靠应用层遍历，无法 SQL 级去重 |
| P5 | **三种事件混合清理** — CleanupProcessed 无法按方向差异化 TTL | `outbox_repository.go:267-278` | 入站事件（需审计保留）被激进清理 |

### 1.2 不在范围内

- Salesforce 回写（已取消）
- Archive 表（当前规模不需要）
- Connector Registry 表（无多实例需求）

## 2. 方案设计

### 2.1 表拓扑

```
_System_OutboxEvent（保留，不改动）
  └─ 职责: 内部 CRM 事件 (record.created/updated/deleted)
  └─ 消费者: ProcessOutbox → EventBus

_System_IntegrationInbound（新建）
  └─ 职责: 外部平台 → CRM 的入站事件
  └─ 生产者: Webhook handler (HubSpot/未来其他平台)
  └─ 消费者: SyncWorker

_System_IntegrationOutbound（新建）
  └─ 职责: CRM → 外部平台的出站变更
  └─ 生产者: PersistenceService (Insert/Update/Delete hooks)
  └─ 消费者: OutboundWorker
```

### 2.2 Inbound 表 Schema

```json
{
  "tableName": "_System_IntegrationInbound",
  "tableType": "system_core",
  "category": "integration",
  "description": "Inbound events from external platforms to NexusCRM",
  "columns": [
    { "name": "Id", "type": "VARCHAR(255)", "primaryKey": true },
    { "name": "Source", "type": "VARCHAR(50)", "nullable": false, "label": "Source Platform" },
    { "name": "ExternalID", "type": "VARCHAR(255)", "nullable": false, "label": "External Record ID" },
    { "name": "ObjectType", "type": "VARCHAR(100)", "nullable": false, "label": "Object Type" },
    { "name": "EventType", "type": "VARCHAR(20)", "nullable": false, "label": "Event Action" },
    { "name": "Payload", "type": "JSON", "nullable": false, "label": "Canonical Event Payload" },
    { "name": "RawPayload", "type": "JSON", "nullable": true, "label": "Original Webhook Body" },
    { "name": "IdempotencyKey", "type": "VARCHAR(255)", "nullable": false, "label": "Dedup Key" },
    { "name": "Status", "type": "VARCHAR(20)", "nullable": false, "default": "'pending'", "label": "Processing Status" },
    { "name": "RetryCount", "type": "INT", "default": "0", "label": "Retry Count" },
    { "name": "ErrorMessage", "type": "TEXT", "nullable": true, "label": "Error Message" },
    { "name": "ProcessedDate", "type": "DATETIME", "nullable": true, "label": "Processed Date" },
    { "name": "CreatedDate", "type": "DATETIME", "nullable": false, "default": "CURRENT_TIMESTAMP" },
    { "name": "LastModifiedDate", "type": "DATETIME", "nullable": false, "default": "CURRENT_TIMESTAMP" }
  ],
  "indices": [
    { "name": "idx_inbound_poll", "columns": ["Status", "CreatedDate"] },
    { "name": "idx_inbound_source", "columns": ["Source", "ObjectType", "CreatedDate"] },
    { "name": "idx_inbound_idempotency", "columns": ["IdempotencyKey"], "unique": true },
    { "name": "idx_inbound_external", "columns": ["Source", "ExternalID"] }
  ]
}
```

**关键设计决策：**
- `Source` — 一等列，替代 EventType 前缀路由（解决 P1）
- `ExternalID` — 一等列，支持 SQL 级去重和索引查询（解决 P4）
- `RawPayload` — 独立列，保留原始 webhook body 用于审计（解决 P5 部分）
- `IdempotencyKey` UNIQUE — 数据库级幂等，`INSERT IGNORE` 即可去重
- `EventType` — 简化为 `create`/`update`/`delete`，不再承担路由职责

### 2.3 Outbound 表 Schema

```json
{
  "tableName": "_System_IntegrationOutbound",
  "tableType": "system_core",
  "category": "integration",
  "description": "Outbound events from NexusCRM to external platforms",
  "columns": [
    { "name": "Id", "type": "VARCHAR(255)", "primaryKey": true },
    { "name": "Destination", "type": "VARCHAR(50)", "nullable": false, "label": "Target Platform" },
    { "name": "ObjectType", "type": "VARCHAR(100)", "nullable": false, "label": "Object Type" },
    { "name": "RecordID", "type": "VARCHAR(255)", "nullable": false, "label": "NexusCRM Record ID" },
    { "name": "EventType", "type": "VARCHAR(20)", "nullable": false, "label": "Event Action" },
    { "name": "Payload", "type": "JSON", "nullable": false, "label": "Record Snapshot" },
    { "name": "OldPayload", "type": "JSON", "nullable": true, "label": "Before-change Snapshot" },
    { "name": "Status", "type": "VARCHAR(20)", "nullable": false, "default": "'pending'", "label": "Processing Status" },
    { "name": "RetryCount", "type": "INT", "default": "0", "label": "Retry Count" },
    { "name": "ErrorMessage", "type": "TEXT", "nullable": true, "label": "Error Message" },
    { "name": "ProcessedDate", "type": "DATETIME", "nullable": true, "label": "Processed Date" },
    { "name": "CreatedDate", "type": "DATETIME", "nullable": false, "default": "CURRENT_TIMESTAMP" },
    { "name": "LastModifiedDate", "type": "DATETIME", "nullable": false, "default": "CURRENT_TIMESTAMP" }
  ],
  "indices": [
    { "name": "idx_outbound_poll", "columns": ["Destination", "Status", "CreatedDate"] },
    { "name": "idx_outbound_record", "columns": ["ObjectType", "RecordID"] }
  ]
}
```

**关键设计决策：**
- `Destination` — 替代 EventBus 内存路由（解决 P3）
- `RecordID` — 一等列，支持按实体聚合和去重
- `OldPayload` — 保留变更前快照，出站 diff 计算用
- 无 `IdempotencyKey` — 出站事件由 CRM 产生，不需要外部去重

### 2.4 Schema 差异对比

| 列 | Inbound | Outbound | 共有？ |
|----|---------|----------|--------|
| Source (来源平台) | ✅ | - | 否 |
| ExternalID | ✅ | - | 否 |
| RawPayload | ✅ | - | 否 |
| IdempotencyKey | ✅ | - | 否 |
| Destination (目标平台) | - | ✅ | 否 |
| RecordID (CRM记录ID) | - | ✅ | 否 |
| OldPayload | - | ✅ | 否 |

7 个方向特有列，0 个真正共有列 — 合表意味着每行一半列为 NULL。

## 3. 接口设计

### 3.1 新增 Repository 接口

> 实际代码: `backend/internal/domain/ports/integration_repositories.go`

```go
// IntegrationInboundRepository handles inbound event persistence.
type IntegrationInboundRepository interface {
    // Enqueue inserts an inbound event. Duplicate IdempotencyKey silently skipped.
    Enqueue(ctx context.Context, event InboundEvent) error

    // GetPending retrieves pending events for a specific source.
    GetPending(ctx context.Context, source string, limit int) ([]InboundEvent, error)

    // Claim locks an event for processing (FOR UPDATE SKIP LOCKED).
    Claim(ctx context.Context, id string) error

    // Complete marks an event as processed.
    Complete(ctx context.Context, id string) error

    // Fail marks an event as failed with error message.
    Fail(ctx context.Context, id string, errMsg string) error

    // IncrementRetry bumps retry count and updates error.
    IncrementRetry(ctx context.Context, id string, errMsg string) error

    // Cleanup deletes processed events older than cutoff.
    Cleanup(ctx context.Context, cutoff time.Time) (int64, error)
}

// IntegrationOutboundRepository handles outbound event persistence.
type IntegrationOutboundRepository interface {
    // Enqueue inserts an outbound event.
    Enqueue(ctx context.Context, event OutboundEvent) error

    // GetPending retrieves pending events for a specific destination.
    GetPending(ctx context.Context, destination string, limit int) ([]OutboundEvent, error)

    // Claim locks an event for processing.
    Claim(ctx context.Context, id string) error

    // Complete marks an event as processed.
    Complete(ctx context.Context, id string) error

    // Fail marks an event as failed.
    Fail(ctx context.Context, id string, errMsg string) error

    // IncrementRetry bumps retry count and updates error.
    IncrementRetry(ctx context.Context, id string, errMsg string) error

    // Cleanup deletes processed events older than cutoff.
    Cleanup(ctx context.Context, cutoff time.Time) (int64, error)
}
```

### 3.2 事件模型

> 实际代码: `shared/pkg/models/integration_events.go`

```go
// InboundEvent represents an event from an external platform.
type InboundEvent struct {
    ID             string     `json:"id"`
    Source         string     `json:"source"`          // "hubspot", "salesforce"
    ExternalID     string     `json:"external_id"`     // external record ID
    ObjectType     string     `json:"object_type"`     // "Contact", "Deal"
    EventType      string     `json:"event_type"`      // "create", "update", "delete"
    Payload        string     `json:"payload"`         // JSON: CanonicalEvent
    RawPayload     string     `json:"raw_payload"`     // JSON: original webhook body
    IdempotencyKey string     `json:"idempotency_key"` // dedup key (UNIQUE)
    Status         string     `json:"status"`          // pending, processing, processed, failed
    RetryCount     int        `json:"retry_count"`
    ErrorMessage   string     `json:"error_message"`
    ProcessedDate  *time.Time `json:"processed_date"`
    CreatedDate    time.Time  `json:"created_date"`
}

// OutboundEvent represents a CRM change to push to an external platform.
type OutboundEvent struct {
    ID            string     `json:"id"`
    Destination   string     `json:"destination"`  // "hubspot"
    ObjectType    string     `json:"object_type"`  // "Contact", "Deal"
    RecordID      string     `json:"record_id"`    // NexusCRM record ID
    EventType     string     `json:"event_type"`   // "create", "update", "delete"
    Payload       string     `json:"payload"`      // JSON: record snapshot
    OldPayload    string     `json:"old_payload"`  // JSON: before-change snapshot (nullable)
    Status        string     `json:"status"`       // pending, processing, processed, failed
    RetryCount    int        `json:"retry_count"`
    ErrorMessage  string     `json:"error_message"`
    ProcessedDate *time.Time `json:"processed_date"`
    CreatedDate   time.Time  `json:"created_date"`
}
```

## 4. 数据流变更

### 4.1 Inbound: 外部 → CRM (before/after)

```
BEFORE:
  Webhook → connector.ParseEvents()
          → outboxRepo.EnqueueWithIdempotency("integration.hubspot.contact.update", CanonicalEvent)
          → SyncWorker.GetPendingEventsByPrefix("integration.hubspot.")
          → json.Unmarshal → CanonicalEvent
          → upsert CRM

AFTER:
  Webhook → connector.ParseEvents()
          → inboundRepo.Enqueue(InboundEvent{Source:"hubspot", ExternalID:"123", ...})
          → SyncWorker.GetPending("hubspot", 100)
          → 直接使用 InboundEvent 结构体（无需反序列化）
          → upsert CRM
```

**改进：**
- ExternalID 是一等列，SQL 级去重（`INSERT IGNORE` on IdempotencyKey）
- Source 是显式枚举，不是字符串前缀
- SyncWorker 不再需要 `GetPendingEventsByPrefix` 的 LIKE 查询

### 4.2 Outbound: CRM → 外部 (before/after)

```
BEFORE:
  PersistenceService.Update()
    → OutboxEvent(record.updated, payload={Record, OldRecord})
    → ProcessOutbox → EventBus.Publish
    → OutboundWorker 内存订阅 EventBus
    → 5秒窗口内累积 → flush to HubSpot
    ⚠️ crash = 数据丢失

AFTER:
  PersistenceService.Update()
    → OutboxEvent(record.updated, payload)  ← 保留，内部事件不变
    → 同时: outboundRepo.Enqueue(OutboundEvent{Destination:"hubspot", RecordID:"xxx"})
    → OutboundWorker 轮询 outboundRepo.GetPending("hubspot", 100)
    → 批量 push to HubSpot
    → outboundRepo.Complete(id)
    ✅ crash-safe: pending 行持久化在 DB
```

**改进：**
- OutboundWorker 从内存订阅改为 DB 轮询，crash-safe（解决 P3）
- ProcessOutbox 不再需要 `skip integration.*`（解决 P1）
- 出站事件有独立的重试/清理策略（解决 P5）

### 4.3 Anti-Loop 机制（不变）

```
HubSpot webhook → InboundEvent → SyncWorker
  → PersistenceService.Insert/Update(ctx with sync_source="hubspot")
    → OutboxEvent(record.created, metadata.sync_source="hubspot")
    → OutboundWorker 检查 payload.Metadata["sync_source"] == "hubspot" → SKIP
```

sync_source 标记机制不变，仍通过 context 传播。

## 5. 代码改动清单

> ✅ 已全部实施，Commit: `7bfc2e3f`

### 5.1 新增文件 (4)

| 文件 | 内容 |
|------|------|
| `shared/pkg/models/integration_events.go` | InboundEvent + OutboundEvent 结构体 |
| `backend/internal/domain/ports/integration_repositories.go` | IntegrationInboundRepository + IntegrationOutboundRepository 接口 |
| `backend/internal/infrastructure/persistence/integration_inbound_repo.go` | Inbound 持久化实现（幂等写入、Source 精确查询、FOR UPDATE SKIP LOCKED） |
| `backend/internal/infrastructure/persistence/integration_outbound_repo.go` | Outbound 持久化实现（Destination 精确查询、OldPayload nullable 处理） |

### 5.2 修改文件 (11)

| 文件 | 变更 |
|------|------|
| `backend/internal/bootstrap/system_tables.json` | 新增 `_System_IntegrationInbound` 和 `_System_IntegrationOutbound` 表定义（含完整索引） |
| `backend/internal/integration/platform/hubspot/sync_worker.go` | `OutboxRepository` → `IntegrationInboundRepository`；`GetPendingEventsByPrefix` → `GetPending("hubspot")`；`UpdateStatus` → `Complete`/`Fail` |
| `backend/internal/integration/platform/hubspot/outbound_worker.go` | `EventBus` 内存订阅 → `OutboundRepository` DB 轮询；删除 `accumulator`/`handleEvent`/`flush`；新增 `runCycle`/`isSyncSourced` |
| `backend/internal/integration/platform/hubspot/routes.go` | 管理端点 `loadMetrics`/`retryDeadLetters` 从 OutboxEvent LIKE 查询改为 IntegrationInbound Source 精确查询 |
| `backend/internal/features/records/persistence_service.go` | 新增 `outboundRepo` 字段 + `SetOutboundRepository` setter + `processPostUpdateHooks` 出站入队（含 anti-loop 检查） |
| `backend/internal/features/records/persistence_insert.go` | insert 事务内新增 outboundRepo.Enqueue（含 `integration_sync_source` anti-loop 检查） |
| `backend/internal/features/records/persistence_delete.go` | delete 事务内新增 outboundRepo.Enqueue（含 anti-loop 检查） |
| `backend/internal/features/events/outbox_service.go` | 删除 `HasPrefix("integration.")` skip 逻辑及 `strings` 导入 |
| `backend/internal/application/services/service_manager.go` | `InfrastructureDeps` 新增 `InboundRepo` + `OutboundRepo` 字段 |
| `backend/internal/application/services/service_manager_init.go` | `SetEnqueueFunc` 改写入 InboundRepo；SyncWorker 构造传 InboundRepo；新增 `SetOutboundRepository` 调用；OutboundWorker 构造传 OutboundRepo |
| `backend/cmd/server/wire.go` | 组合根注入 `NewIntegrationInboundRepository` + `NewIntegrationOutboundRepository` |

### 5.3 自动生成文件 (codegen 产出)

| 文件 | 内容 |
|------|------|
| `shared/pkg/constants/z_generated_tables.go` | `TableIntegrationInbound` + `TableIntegrationOutbound` 常量 |
| `shared/pkg/constants/z_generated_fields.go` | 两张新表的字段常量 |
| `shared/pkg/constants/z_generated_columns.go` | 两张新表的列列表 |
| `shared/pkg/models/z_generated.go` | 两张新表的 SObject 转换 |
| `frontend/src/generated-schema.ts` | 前端 Schema 定义同步 |

### 5.4 不改动

| 文件 | 原因 |
|------|------|
| `integration/connector.go` | Connector 接口不变 |
| `integration/canonical_event.go` | CanonicalEvent 结构不变（仍用于 Payload 序列化） |
| `integration/worker.go` | BackgroundWorker 接口不变 |
| `infrastructure/persistence/outbox_repository.go` | OutboxRepository 保留，服务内部事件 |

## 6. 迁移策略

> ✅ 实施采用一步切换（非双写），因为当前只有 1 个集成且无生产流量。

### 6.1 实际实施步骤

```
一步切换（已完成）:
  1. system_tables.json 添加新表定义 → Bootstrap 自动建表
  2. Webhook handler 直接写 IntegrationInbound（不再写 OutboxEvent）
  3. PersistenceService hooks 直接写 IntegrationOutbound（不再依赖 EventBus）
  4. SyncWorker 直接读 IntegrationInbound
  5. OutboundWorker 直接轮询 IntegrationOutbound
  6. 删除 ProcessOutbox 的 HasPrefix skip 逻辑
```

### 6.2 原双写方案（备用，适用于有生产流量时）

```
Phase 1: 建表 + 双写
  1-4. Webhook handler 同时写 OutboxEvent + IntegrationInbound
       PersistenceService hooks 同时写 OutboxEvent + IntegrationOutbound

Phase 2: 切换消费者
  5-8. SyncWorker/OutboundWorker 切到新表，观察 24h

Phase 3: 停止双写
  9-11. 停止向 OutboxEvent 写 integration.* 事件
```

### 6.3 回滚方案

- `git revert 7bfc2e3f` 即可恢复到旧架构
- 新表保留在数据库中不影响旧代码（无消费者）

## 7. 分表触发条件（ADR）

当满足以下**任一**条件时，考虑进一步拆分（如按 Source/Destination 独立表）：
- 接入第 3 个集成平台
- 单张 Inbound/Outbound 表行数超过 100 万
- 监控发现跨 Source/Destination 的 UPDATE 锁竞争

## 8. 实施备注

### 8.1 与原方案的偏差

| 原方案 | 实际实施 | 原因 |
|--------|---------|------|
| Models 放在 `backend/internal/domain/models/` | 放在 `shared/pkg/models/` | 与 `OutboxEvent` 保持一致，跨模块共享 |
| 双写迁移（三阶段） | 一步切换 | 只有 1 个集成且无生产流量，双写增加不必要复杂度 |
| `Destination` 由 `IntegrationDestinationResolver` 动态提供 | 硬编码 `"hubspot"` | MVP 阶段只有一个集成，YAGNI；设计预留了 `SetOutboundRepository` setter 模式便于后续扩展 |
| `routes.go` 不在改动清单中 | 实际修改了 `routes.go` | `loadMetrics` 和 `retryDeadLetters` 引用了已删除的 `outboxEventPrefix` 常量 |

### 8.2 Anti-Loop 机制验证

Anti-loop 通过 3 层保护（未改变）：

1. **Context 传播**: `SyncWorker` 注入 `integration_sync_source=hubspot` 到 context
2. **Metadata 标记**: `PersistenceService` 读取 context 并写入 `payload.Metadata["sync_source"]`
3. **出站检查**: 
   - `PersistenceService` hooks: `syncSource == ""` 时才入队（不从 integration sync 写入）
   - `OutboundWorker.isSyncSourced()`: 解析 payload JSON 二次检查

### 8.3 验证结果

```
go build ./...       ✅ 通过
go vet ./...         ✅ 通过
go test -race ./...  ✅ 全部通过（含 hubspot 包测试）
codegen              ✅ 生成新常量
```

## 9. 辩论过程摘要（存档）

本方案经历两轮 Agent 辩论（共 6 回合）：

**第一轮（拆分派 vs 演进派）：** 从三表方案收敛到双表（去掉 Archive），确认 idempotency_key 单列 UNIQUE、参数化 Repository、source 列替代前缀路由。

**第二轮（务实派 vs 结构派）：** 务实派主张单表 + source 列足够，结构派主张双表。最终结构派承认当前规模单表可行，但用户决定：**现在是实施双表代价最低的时机**（只有 1 个集成，迁移成本最小），确认执行双表方案。