# 计量计价重构 — 分步实施计划

> 基于「计量+计价新方案（分离表）」和「计价四阶段插件设计方案」两份设计文档。
> 目标：将现有 `measure.go` 中 `costUserRawUnitCost → costUnitUsage → txCost` 单体计价链路，
> 重构为四阶段插件化 pipeline 架构，并实现计量与计价的表级分离。

## 当前状态 (2026-07-24)

| Step | 状态 | 关键文件 |
|------|------|---------|
| 0 — Pipeline 框架骨架 | ✅ 完成 | `pipeline/{context,interfaces,registry,outcome,detail,pipeline,assembler}.go` |
| 1 — DB 表 + config flags | ✅ 完成 | `entity/{metering_event,pricing_event}.go`, `migrate_record167.go`, `config.go` |
| 2 — Stage 1 插件 | ✅ 完成 | `adjusters/{normal_unit_price,threshold_unit_price,interval_unit_price,normal_usage}.go` |
| 3 — Stage 3 插件 | ✅ 完成 | `adjusters/binded_package_discount.go` + `pipeline/provider.go` |
| 4 — Stage 4 插件 | ✅ 完成 | `adjusters/tiered_billing.go` |
| 5 — pipeline.Run 核心 | ✅ 完成 | `pipeline/pipeline.go` (runPricing + runTakeover + settleCredits + prepareMergeTarget + writeCostResult) |
| 6 — AssemblePipeline 路由 | ✅ 完成 | `pipeline/assembler.go` |
| 7a — SQS 双写 | ✅ 完成 | `measure.go::writeMeteringEventDual` |
| 7b — Pricing Worker | ✅ 完成 | `workers/pricing_worker.go` + `pipeline_provider.go` + worker 注册 |
| 7c — 核心单测 | ✅ 完成 | `pipeline_test.go` (8 tests) + `adjusters/adjusters_test.go` (7 tests) — 27 subtests 全 PASS |
| 8 — Reconciliation Worker | ✅ 完成 | `workers/pricing_reconciliation_worker.go` + Prometheus metrics |
| 9 — Phase 2 灰度接管 | 🔲 待实施 | per-org `UsePipelineResult=true` |
| 10 — Phase 3 下线 | 🔲 待实施 | 旧路径删除 + 新插件 |

> Branch: `refactor/pricing-pipeline-plugin`
> Build: `go build ./...` ✅ | Tests: `go test ./model/billing/pipeline/...` ✅ (27 subtests)

## 现有代码 → 新架构映射速查

| 现有代码 | 位置 | 新架构对应 | 说明 |
|---------|------|-----------|------|
| `costUserRawUnitCost` | measure.go:913 | pipeline 调用方 (§4.4.7) | 数据查询下沉到插件；入口接收 meteringEvent + pricingEvent + unitInfo |
| `costUnitUsage` | measure.go:369 | `pipeline.Run` (§4.4.3) | 算价主流程 → Stage 1-4 线性执行 |
| `txCost` | measure.go:467 | Run 事务内 (A + Stage1-4 + E + F) | Stage 1-3 MVCC 快照读，事务范围等宽 |
| `getRawPriceFromSku` | measure.go:346 | Stage 1 插件 Propose | switch-case → Adjuster |
| `dealThresholdPrice` | measure.go:276 | `ThresholdUnitPriceAdjuster` | |
| `dealIntervalPrice` | measure.go:302 | `IntervalUnitPriceAdjuster` | |
| `GetUseDiscount` | measure.go:780 | Stage 3 插件内部自查 | |
| `ListBindedPackages` | measure.go:688 | Stage 3 插件内部自查 | |
| `getUseBindedPackage` | measure.go:189 | Stage 3 `BindedPackageDiscountAdjuster` | 6 级优先级选套餐 |
| `getOrCreateMergedUnitCost` | measure.go:980 | `prepareMergeTarget` (事务外) | |
| `labelMatch` (tiered) | measure.go:416 | `hasTieredBilling` + Stage 4 插件 | |
| `TieredBilling.Accumulate` | measure.go:523 | `TieredBillingCostPriceAdjuster.Propose` | |

## 实施阶段

---

### Step 0: 搭建 Pipeline 框架骨架（不接入热路径）

**目标**：创建所有核心类型、接口、注册表、组装器和执行器的空壳，编译通过但零生产影响。

**交付物**：

```
model/billing/pipeline/
├── context.go       # PricingContext + 分阶段 Result struct + AuditEntry
├── interfaces.go    # ProposalKind + 4 种 Adjuster 接口 + Proposal struct + ShortCircuitInfo
├── registry.go      # PluginRegistry 单例 + Register 方法
├── assembler.go     # AssemblePipeline 空壳（暂时 hardcode 返回空 pipeline）
├── pipeline.go      # PricingPipeline.Run 空壳（暂时返回 OutcomeNormal）
├── outcome.go       # PricingOutcome + OutcomeKind 枚举
├── detail.go        # PricingDetail struct + buildPricingDetail
└── adjusters/       # 空目录，后续填充
```

**具体工作**：

1. 定义 `PricingContext` struct（§4.1）
   - MeteringEntity / PricingEntity / UnitInfo / Tx
   - UnitPrice / Usage / Discount / CostPrice 四个 Result struct 指针
   - PipelineSettleCreditsAndMerge / AuditTrail

2. 定义四类插件接口（§4.2.3）
   - `UnitPriceAdjuster` / `UsageAdjuster` / `DiscountAdjuster` / `CostPriceAdjuster`
   - 统一 `Propose(ctx *PricingContext) (*XxxProposal, error)` + `Name() string` 签名

3. 定义 `ProposalKind` 枚举 + `ProposalNormal/NotApplicable/ShortCircuit`（§4.2.2）

4. 定义各阶段 Proposal struct（§4.2.3）
   - `UnitPriceProposal` / `UsageProposal` / `DiscountProposal` / `CostPriceProposal`

5. 定义 `ShortCircuitInfo`（§4.2.4）

6. 定义 `PricingOutcome` + `OutcomeKind`（§4.4.1）

7. 定义 `PluginRegistry` 单例 + `PricingPipeline` struct（§4.3.2）
   - Registry 是进程级 `map[type_key]Adjuster`
   - Pipeline 持有四组 `[]Adjuster` 切片

8. 定义 `PricingDetail` JSON struct（§4.4.4 detail.go 部分）

**验证标准**：`go build ./model/billing/pipeline/...` 编译通过。

**不包含**：任何 DB 迁移、任何热路径改动、任何插件实现。

---

### Step 1: 新增 DB 表 + Entity（metering_events / pricing_events）

**目标**：创建两张新表的 entity 和 migration，不写入数据。

**交付物**：

1. `db/entity/metering_event.go` — `MeteringEvent` entity
   - 字段按文档 §3 的 `metering_events` 表定义
   - 生成 queryset 代码（`goqueryset` 或手写）
   - 对应 DDL：

   ```sql
   CREATE TABLE metering_events (
       id              BIGINT       NOT NULL AUTO_INCREMENT,
       cost_id         VARCHAR(128) NOT NULL,
       tenant_id       INT UNSIGNED NOT NULL,
       unit            VARCHAR(64)  NOT NULL,
       cost            INT          NOT NULL,
       cost_time_stamp BIGINT       NOT NULL,
       labels          JSON         NOT NULL,
       system_labels   JSON         NULL,
       status          VARCHAR(32)  NOT NULL DEFAULT 'to_price',
       created_at      TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
       PRIMARY KEY (id),
       UNIQUE KEY uk_cost_id (cost_id),
       KEY idx_tenant_time (tenant_id, cost_time_stamp),
       KEY idx_status_id (status, id)
   );
   ```

   `system_labels`（`map[string]string` JSON 列）用于系统内部标记，
   如 `{"dryrun":"true"}`、`{"measure_type":"secondary"}`、`{"testing":"true"}`，
   与资源匹配用的 `labels` 隔离，不影响 tiered billing / binded package 的 label 匹配。

2. `db/entity/pricing_event.go` — `PricingEvent` entity
   - 字段按文档 §4 的 `pricing_events` 表定义
   - 含状态枚举 `PricingStatus_ToPrice / Pricing / Priced / Failed`
   - `ConsumedSource` 常量 `Credits / Invoice / None`
   - `IsEffective bool` + `MergedCostID *string`

3. Migration（gormigrate，遵守 AGENTS.md 的 footgun 规则）
   - `migrate/` 下新增 migration 记录
   - 每条 DDL 带 `information_schema` 幂等检查
   - 包含所有索引（`uk_cost_id`, `idx_priced_id`, `idx_tenant_time`, `idx_cost_effective`, `idx_status_id`, `idx_merged_cost_id`, `idx_tenant_time_effective`）

4. 新增 config flag
   - `EnableMeteringEvents`（热路径写 metering_events 开关，默认 `false`）
   - `EnablePricingPipeline`（pipeline 开关，默认 `false`）
   - `UsePipelineResult`（接管模式开关，默认 `false`）

**验证标准**：
- `go build ./...` 通过
- Migration 在 dev DB 上可正反执行
- `make test` 通过（现有测试不受影响）

---

### Step 2: Stage 1 插件 — 调单价（Normal / Threshold / Interval）

**目标**：将现有 `getRawPriceFromSku` 的三种 price_type 平移为插件，行为不变。

**交付物**：

```
model/billing/pipeline/adjusters/
├── normal_unit_price.go       # NormalUnitPriceAdjuster
├── threshold_unit_price.go    # ThresholdUnitPriceAdjuster（封装 dealThresholdPrice）
└── interval_unit_price.go     # IntervalUnitPriceAdjuster（封装 dealIntervalPrice）
```

**具体工作**：

1. `NormalUnitPriceAdjuster.Propose`
   - 返回 `ProposalNormal`，`AdjustedPrice = ctx.UnitInfo.Price`

2. `ThresholdUnitPriceAdjuster.Propose`
   - 封装 `dealThresholdPrice` 逻辑（measure.go:276）
   - 读 `unit.PriceExpression` 中 `multiple_factor_label` + `multiple_factor_threshold`
   - 从 `ctx.MeteringEntity.Labels` 读 label 值
   - `factor = max(0, value - threshold)`, `AdjustedPrice = Unit.Price * factor`
   - label 缺失 → `ProposalShortCircuit{stage:"stage1", FallbackStatus:"Failed"}`

3. `IntervalUnitPriceAdjuster.Propose`
   - 封装 `dealIntervalPrice` 逻辑（measure.go:302）
   - 读 labels 中的 `intervalMin/Max/Value`，clamp 到 `[min, max]`
   - `AdjustedPrice = Unit.Price * clampedFactor`

4. 在 `registry.go` 中注册三种插件

5. 编写单元测试（table-driven，覆盖 normal / threshold / interval / label-missing 场景）

**验证标准**：
- 单元测试覆盖所有分支
- 插件不触碰 `PricingEntity`，只返回 Proposal

---

### Step 3: Stage 3 插件 — 调折扣 + 套餐选择

**目标**：封装 `GetUseDiscount` + `ListBindedPackages` + `getUseBindedPackage` 为 `BindedPackageDiscountAdjuster`。

**交付物**：

```
model/billing/pipeline/adjusters/
└── binded_package_discount.go  # BindedPackageDiscountAdjuster
```

**具体工作**：

1. `BindedPackageDiscountAdjuster.Propose`
   - 内部调用 `GetUseDiscount(ctx.MeteringEntity, ctx.UnitInfo, isBeta)`
   - 内部调用 `ListBindedPackages(ctx.MeteringEntity, ctx.UnitInfo, isBeta)`
   - 内部调用 `getUseBindedPackage(tenantID, packages, rawPrice, discount, dryrun)`
   - `rawPrice = ctx.UnitPrice.AdjustedPrice`（来自 Stage 1 结果）
   - 如果 `selectedPackage == nil` → `ProposalShortCircuit{stage:"stage3", FallbackStatus:"DONE", Internal:true}`
   - 正常 → `ProposalNormal{Discount, DiscountID, SelectedPackage, ConsumedSource}`

2. `ConsumedSource` 判断逻辑（与现有一致）
   - `selectedPackage.RemainingCredits > 0` → `Credits`
   - 否则 → `Invoice`

3. 编写单元测试（mock discount / binded_packages 查询）

**验证标准**：
- 单元测试验证套餐选择优先级不变
- ShortCircuit 路径正确触发

**注意**：此插件依赖 `MeasureModel` 的方法（`GetUseDiscount` / `ListBindedPackages`），
需要通过 `PricingContext` 或闭包传入 model 引用。

---

### Step 4: Stage 4 插件 — TieredBilling

**目标**：封装 `TieredBilling.Accumulate` 为 `TieredBillingCostPriceAdjuster`。

**交付物**：

```
model/billing/pipeline/adjusters/
└── tiered_billing.go  # TieredBillingCostPriceAdjuster
```

**具体工作**：

1. `TieredBillingCostPriceAdjuster.Propose`
   - 自查 tiered_billings 配置（labelMatch）
   - 事务内 `TieredBilling.Accumulate(ctx.Tx, ...)`（read-modify-write 累加器）
   - 超出 target → 截断价 = `target - accumulated + base`
   - 正常 → `ProposalNormal{AdjustedPrice = baseCostPrice}`（无截断）
   - 无匹配 tiered → `ProposalNotApplicable`

2. 编写单元测试

---

### Step 5: 实现 Pipeline.Run 核心执行逻辑

**目标**：实现 Stage 1→2→3→4 线性执行 + 选最优 + 事务控制。

**交付物**：

完善 `pipeline.go`：

1. `runStage1to3`（§4.4.3）
   - Stage 1: 遍历 `unitPriceAdjusters`，收 Proposal，选最低价
   - Stage 2: 遍历 `usageAdjusters`，收 Proposal，选最低用量
   - Stage 3: 遍历 `discountAdjusters`，收 Proposal
   - 短路 → `fallbackWrite` + 返回 `OutcomeShortCircuit`
   - 计算 `baseCostPrice = skuPrice × adjustedUsage × discount`
   - 写 `pricing.RawPrice = skuPrice × adjustedUsage`

2. `runStage4`（§4.4.3）
   - 遍历 `costPriceAdjusters`，收 Proposal，选最低总价
   - 计算 `Adjustment = bestTotal - baseCostPrice`

3. `runPricing`（worker 模式，§4.4.3）
   - 无事务，只算价 + 写 pricing_events（`is_effective=false`）
   - `PipelineSettleCreditsAndMerge=false` 时走此路径

4. `runTakeover`（接管模式，§4.4.3）
   - 全部 Stage 在同一事务内
   - A段: `lockPricingEvent`（FOR UPDATE pricing_event status=to_price）
   - Stage 1-3（MVCC 快照读）
   - Stage 4（插件可 FOR UPDATE）
   - E段: `settleCredits`（读 remaining_credits FOR UPDATE + 扣减）
   - F段: `writeCostResult`（target FOR UPDATE + merge + pricing_event → priced/effective）
   - `PipelineSettleCreditsAndMerge=true` 时走此路径

5. `fallbackWrite`（§4.4.4）
   - 短路兜底：写 pricing_event status=failed / internal=true / consumed_source=none

6. `settleCredits`（§4.4.4）
   - 读 `binded_packages.remaining_credits` FOR UPDATE
   - 耗光 → `RemeasureError`（事务回滚）
   - 扣减 `remaining_credits -= costPrice`

7. `writeCostResult`（§4.4.4）
   - target FOR UPDATE + merge 累加（Cost/RawPrice/CostPrice/Adjustment）
   - 或创建新 merged 行
   - 更新 pricing_event: status=priced, is_effective=true

8. `prepareMergeTarget`（§4.4.4）
   - 封装 `getOrCreateMergedUnitCost` 逻辑
   - **注意 FORCE INDEX** `tenant_timestamp_status_cluster`（STORED GENERATED column）

**验证标准**：
- 单元测试覆盖 Normal / ShortCircuit / Remeasure / Skipped 四种 Outcome
- worker 模式和接管模式均测试

---

### Step 6: 实现 AssemblePipeline 路由组装

**目标**：根据配置表动态组装 pipeline。

**交付物**：

完善 `assembler.go`：

1. Stage 1 路由
   - 查 `tenant_units` 表（需新建 entity + migration，见下方 Step 6a）
   - 无匹配 → 回退查 `units.PriceType`（兼容老数据）
   - 加载对应 `UnitPriceAdjuster`

2. Stage 2 路由
   - 查 `tenant_usages` 表（需新建 entity + migration）
   - 加载对应 `UsageAdjuster`（当前只有 NormalUsageAdjuster，Stage 2 预留）

3. Stage 3 路由
   - 固定加载 `BindedPackageDiscountAdjuster`

4. Stage 4 路由
   - 查 `tiered_billings` 是否有匹配记录
   - 有则加载 `TieredBillingCostPriceAdjuster`

**Step 6a（可选并行）**：新增配置表

```
tenant_units + tenant_unit_filter（§3.3）
tenant_usages + tenant_usage_filter（§3.4）
```

- Migration（gormigrate + information_schema 幂等检查）
- Entity + QuerySet
- 数据迁移脚本：从 `units.PriceType` → `tenant_units.price_type`（tenant_id=0 全局默认）

**验证标准**：
- 路由正确匹配到对应插件
- 无配置时回退到 normal

---

### Step 7: 接入热路径（Phase 1 双写模式）

**目标**：热路径新增 `INSERT metering_events`，不影响旧路径。Pricing Worker 离线消费。

**交付物**：

1. `model/billing/measure.go` — SQS handler / `procRawUnitCost` 内新增
   - `saveMeteringEvent(metering)` — feature flag `EnableMeteringEvents`
   - 写失败只 log，不影响热路径

2. `cmd/worker/main.go` — 新增 Pricing Worker goroutine
   - 扫 `metering_events WHERE priced=0 ORDER BY id ASC`
   - 创建 `pricing_events`（status=to_price）
   - `AssemblePipeline + pipeline.Run`（worker 模式）
   - `recoverStuck` 定时重置 pricing → to_price（超 10min）

3. `createPricingEvent(metering)` 辅助函数

4. Pricing Worker 监控指标
   - `PricingWorkerPendingGauge` / `PricingWorkerErrorCounter`
   - `PricingWorkerFailedGauge` / `PricingWorkerStuckGauge`

**验证标准**：
- `EnableMeteringEvents=false` 时零影响
- `EnableMeteringEvents=true` 时 metering_events 正确写入
- Pricing Worker 正确消费并产出 pricing_events（is_effective=false）

---

### Step 8: 双写对比 + Reconciliation Worker

**目标**：逐事件对比旧路径 `raw_user_unit_costs` vs 新流程 `pricing_events`。

**交付物**：

1. `model/billing/pipeline/compare.go`
   - `CompareResults(ruuc *RawUserUnitCost, pe *PricingEvent) *ComparisonResult`
   - 逐字段对比：cost_price / raw_price / consumed_source / pricing_detail

2. Reconciliation Worker goroutine
   - 每 10 分钟一轮，覆盖最近 1 小时
   - `JOIN raw_user_unit_costs × pricing_events ON cost_id`
   - mismatch 写入 `pipeline_comparison_log` 表
   - Stage 2 智能跳过（查 tenant_usages 配置）

3. 监控指标
   - `ReconciliationMismatchGauge` / `PricingEventCoverageGauge`
   - 告警阈值：mismatch > 0.01% 或 coverage < 99.9%

**验证标准**：
- 在 staging 环境运行，mismatch rate < 0.01% 持续 2 周

---

### Step 9: Phase 2 — 分 org 灰度接管

**目标**：per-org `UsePipelineResult=true`，新流程接管 credits 扣减 + merge。

**交付物**：

1. `measure.go` 的 `costUserRawUnitCost` 改造
   - `UsePipelineResult=true` 时走 `pipeline.Run`（接管模式）
   - 旧路径对该 org `takeOver=true` → `computeOldWay` 跳过落库
   - 互斥保证：两条路径不会同时扣 credits / 合并

2. per-org 灰度配置（config 文件或 DB 表）

3. 灰度上线：小 org → 中 org → 大 org，每批观察 3 天

**验证标准**：
- 接管 org 的 credits 扣减 + merge 无异常
- 回滚：per-org `UsePipelineResult=false`（秒级生效）

---

### Step 10: Phase 3 — 旧路径下线 + 新需求插件

**目标**：全量接管后删除旧路径代码，新增业务插件。

**交付物（按需迭代）**：

1. 旧路径代码删除 / feature gate 永久关闭
2. `raw_user_unit_costs` 停写
3. 新增插件（M3 里程碑）：
   - `VolumeTieredUnitPriceAdjuster`（Stage 1，阶梯定价）
   - `FreeTierUsageAdjuster`（Stage 2，免费额度）
   - `ActiveResourceHourAdjuster`（Stage 2，活跃集群小时）
   - `CostCapCostPriceAdjuster`（Stage 4，费用封顶）
   - `PercentageOfCostAdjuster`（Stage 4，百分比计费）

---

## 文件清单（新建）

```
model/billing/pipeline/
├── context.go                          # Step 0
├── interfaces.go                       # Step 0
├── registry.go                         # Step 0
├── assembler.go                        # Step 0 (空壳) → Step 6 (填充)
├── pipeline.go                         # Step 0 (空壳) → Step 5 (填充)
├── outcome.go                          # Step 0
├── detail.go                           # Step 0
├── compare.go                          # Step 8
└── adjusters/
    ├── normal_unit_price.go            # Step 2
    ├── threshold_unit_price.go         # Step 2
    ├── interval_unit_price.go          # Step 2
    ├── normal_usage.go                 # Step 6 (Stage 2 默认，预留)
    ├── binded_package_discount.go      # Step 3
    ├── tiered_billing.go               # Step 4
    ├── volume_tiered.go                # Step 10 (M3)
    ├── free_tier.go                    # Step 10 (M3)
    ├── active_resource_hour.go         # Step 10 (M3)
    ├── cost_cap.go                     # Step 10 (M3)
    └── percentage_of_cost.go           # Step 10 (M3)

db/entity/
├── metering_event.go                   # Step 1
├── pricing_event.go                    # Step 1
├── tenant_unit.go                      # Step 6a
├── tenant_usage.go                     # Step 6a
└── pipeline_comparison_log.go          # Step 8

migrate/
└── (migration records)                 # Step 1, 6a, 8

worker/pricing/
├── pricing_worker.go                   # Step 7
└── pricing_worker_test.go              # Step 7
```

## 关键注意事项

1. **gormigrate footgun**：每条 migration 必须 `information_schema` 幂等检查，不能用 `IF NOT EXISTS`。
2. **FORCE INDEX**：`getOrCreateMergedUnitCost` 中的 `tenant_timestamp_status_cluster` 索引必须保留 FORCE INDEX（STORED GENERATED column）。
3. **measure-path SQL 纪律**：新引入的查询如影响 measure path QPS，必须加 feature flag + EXPLAIN 测试 + benchmark。
4. **money 计算**：全程使用 `shopspring/decimal`，禁止 `float64`。
5. **插件不可变性**：上线后只替换不修改，配合 `pricing_events.pricing_detail` 记录快照。
6. **事务边界**：全部 Stage 1-4 在同一事务内；Stage 1-3 是 MVCC 快照读；A 段锁 pricing_event，target FOR UPDATE 推迟到 F 段。
7. **双写安全**：Phase 1 期间 pricing_events 的 `is_effective=false`，不参与 merge，不影响 user_unit_costs。
8. **回滚路径**：每一步都可通过 feature flag 秒级回滚，不需要 revert 代码。