Files
junhong_cmp_fiber/openspec/specs/polling-task-handlers/spec.md
huang 8bf1282378
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m24s
归档: add-polling-card-status-with-verbose-log,同步主规范
change 目录归档至 openspec/changes/archive/2026-04-15-add-polling-card-status-with-verbose-log/
新建主规范 openspec/specs/polling-card-status-task/spec.md(6个需求)
追加 verbose log 需求至 openspec/specs/polling-task-handlers/spec.md(4个新需求)
protect handler 规范更新字段名:network_status → network_status_at_check + action_taken

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
2026-04-15 12:23:05 +08:00

350 lines
20 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
### Requirement: polling_handler.go 拆分为 4 个专注 Handler
`internal/task/polling_handler.go`1360行拆分为 4 个职责单一的文件,每个文件只负责一种任务类型的数据采集,停复机决策统一委托给 `StopResumeService.EvaluateAndAct()`
**文件目标行数**
| 文件 | 职责 | 目标行数 |
|------|------|---------|
| `polling_base.go` | 共享基类(并发/缓存/重入队) | < 150行 |
| `polling_realname_handler.go` | 实名状态采集 | < 200行 |
| `polling_carddata_handler.go` | 流量数据采集 | < 300行 |
| `polling_package_handler.go` | 套餐数据采集 | < 200行 |
| `polling_protect_handler.go` | 保护期一致性检查 | < 150行 |
**删除**:拆分完成后,`internal/task/polling_handler.go` 整体删除。
**Handler 边界原则**
- ✅ 允许:调 Gateway 采集数据、写 StoreDB更新、调 StopResumeService.EvaluateAndAct()
- ❌ 禁止:直接 `h.db.Model()`(绕过 Store 层)、停复机判断逻辑、直接调 Gateway 停复机接口
#### Scenario: Asynq 任务类型常量不变(向后兼容)
- **GIVEN** 现有 Asynq Worker 已注册 4 种任务类型常量
- **WHEN** 拆分后注册 4 个新 Handler
- **THEN** 任务类型常量名称完全不变(`TaskTypeRealnameCheck``TaskTypeCarddataCheck` 等);已在 Asynq 队列中的待处理任务无需清空,新 Handler 直接接管处理
#### Scenario: 4 个 Handler 并行处理不干扰
- **GIVEN** 同时有 realname、carddata、package、protect 四种任务在处理
- **WHEN** 各 Handler 独立执行
- **THEN** 各 Handler 使用独立的并发锁(`PollingBase.acquireConcurrency` 按任务类型隔离realname 任务的并发数不占用 carddata 任务的并发配额
---
### Requirement: 共享基类 PollingBase
新建 `internal/task/polling_base.go`,提供 `PollingBase` 结构体,供 4 个 Handler 组合使用。
**提取的公共方法**(原来在 `polling_handler.go` 中重复出现):
- `acquireConcurrency(taskType) bool`:获取并发控制信号量
- `releaseConcurrency(taskType)`:释放并发控制信号量
- `getCardWithCache(ctx, cardID) (*model.IotCard, error)`:带 Redis 缓存的卡查询
- `updateCardCache(ctx, card)`:更新 Redis 卡信息缓存
- `requeueCard(ctx, cardID, taskType, interval)`:按间隔重新入队(调 `PollingQueueManager.Requeue`
- `getMatchedPollingInterval(card) time.Duration`:获取该卡匹配的轮询间隔
#### Scenario: 缓存命中减少 DB 压力
- **GIVEN** cardID=100 的卡信息已缓存在 RedisTTL 5分钟
- **WHEN** `polling_realname_handler.go` 调用 `base.getCardWithCache(ctx, 100)`
- **THEN** 直接从 Redis 读取,不查询 DB缓存命中后更新 TTL滑动窗口
#### Scenario: 缓存未命中回源 DB
- **GIVEN** cardID=200 的卡信息不在 Redis 缓存中
- **WHEN** 任一 Handler 调用 `base.getCardWithCache(ctx, 200)`
- **THEN** 查询 DB将结果写入 RedisTTL 5分钟返回卡信息
#### Scenario: 并发控制防止过载——并发满时必须 requeue
- **GIVEN** `carddata` 任务最大并发数配置为 50
- **WHEN** 同时有 60 个 carddata 任务尝试执行
- **THEN** 前 50 个获取到信号量正常执行;后 10 个 `acquireConcurrency` 返回 false**后 10 个任务调 `requeueCard(ctx, cardID, taskType, time.Now())` 立即重入队**,记录 Debug 日志「并发数已满,已重新入队: cardID=xxx」返回 nilAsynq 不报错,不重试)
> **⚠️ 正确性约束**Lua 脚本原子出队时卡已从 Redis Sorted Set 中删除。若并发满时直接 return nil 而不 requeue该卡将永久消失不再被轮询。所有 Handler 必须在 `acquireConcurrency` 返回 false 时先调 `requeueCard` 再返回。
---
### Requirement: polling_realname_handler.go——实名数据采集
**文件**`internal/task/polling_realname_handler.go`< 200行
**构造函数依赖**(通过构造函数注入,禁止全局变量):
```go
func NewPollingRealnameHandler(
base *PollingBase,
gateway GatewayClient,
iotCardStore IotCardStore,
queueClient QueueClient, // 用于触发首次实名激活任务
) *PollingRealnameHandler
```
**方法列表**
- `Handle(ctx, task) error`Asynq 任务入口
- `processCard(ctx, cardID) error`:核心处理逻辑
**处理流程**
1.`base.acquireConcurrency("realname")`,获取失败则跳过
2.`base.getCardWithCache(ctx, cardID)` 获取卡信息
3. 调 Gateway 查询实名状态
4. 写 Store更新 `real_name_status`
5. 若实名状态变为已实名0→1
a. 入队首次实名激活 Asynq 任务(`triggerFirstRealnameActivation`
b. **调 `stopResumeService.EvaluateAndAct(ctx, card)` 触发复机判断**(卡可能因 `not_realname` 停机,需立即复机)
6.`base.requeueCard(ctx, cardID, "realname", interval)` 重新入队
7.`base.releaseConcurrency("realname")`
#### Scenario: 实名状态由未实名变为已实名触发激活和复机
- **GIVEN** cardID=100`real_name_status=0`(未实名),且卡因 `not_realname` 处于停机状态(`network_status=0, stop_reason='not_realname'`
- **WHEN** Gateway 返回实名已完成,`HandleRealnameCheck` 执行
- **THEN** Store 更新 `real_name_status=1`;入队首次实名激活 Asynq 任务;**立即调用 `EvaluateAndAct` 检测复机条件**;若有有效套餐且流量未耗尽,发起 Gateway 复机调用DB 更新 `network_status=1, stop_reason=''`;记录日志「卡实名状态变更: cardID=100, 0→1触发激活任务和复机评估」
#### Scenario: 实名状态未变化不触发激活
- **GIVEN** cardID=200`real_name_status=1`已实名Gateway 返回仍已实名
- **WHEN** `HandleRealnameCheck` 执行
- **THEN** Store 不执行更新(无变化);不入队激活任务;调 `requeueCard` 按正常间隔重新入队
#### Scenario: realname Handler 仅在实名 0→1 时调用 EvaluateAndActS1 修复)
- **GIVEN** `polling_realname_handler.go` 代码 Review
- **WHEN** 检查文件内容
- **THEN** 文件中**不出现**无条件的 `stopCard``resumeCard` 等停复机操作;但**允许且必须**在实名状态由 0→1 时调用 `stopResumeService.EvaluateAndAct(ctx, card)` 触发复机判断;原因:若卡因 `not_realname` 停机,实名完成后应立即复机,不能等下一个 carddata/package 轮询周期(可能长达 1 小时)
---
### Requirement: polling_carddata_handler.go——流量数据采集
**文件**`internal/task/polling_carddata_handler.go`< 300行
**构造函数依赖**
```go
func NewPollingCarddataHandler(
base *PollingBase,
gateway GatewayClient,
iotCardStore IotCardStore,
packageStore PackageUsageStore,
usageService UsageService, // DeductDataUsage
stopResumeService StopResumeServiceInterface,
) *PollingCarddataHandler
```
**方法列表**
- `Handle(ctx, task) error`Asynq 任务入口
- `processCard(ctx, cardID) error`:核心处理逻辑
- `collectUsageData(ctx, card) (*UsageData, error)`:调 Gateway 获取流量增量
**处理流程**
1. 获取并发控制
2. 调 Gateway 查询流量增量
3. 写 Store更新 `data_usage_mb`
4.`usageService.DeductDataUsage()`(套餐流量扣减计算)
5.`stopResumeService.EvaluateAndAct(ctx, card, ...)`(停复机决策)
6. 重新入队,释放并发控制
#### Scenario: 流量更新后委托停复机决策
- **GIVEN** cardID=100流量更新后 `data_usage_mb` 达到 `data_limit_mb`
- **WHEN** `HandleCarddataCheck` 执行完流量写入
- **THEN** 调用 `stopResumeService.EvaluateAndAct(ctx, card, "iot_card", 100)`;由 StopResumeService 决定是否停机Handler 不包含 `if usage >= limit { stop() }` 这类判断代码
#### Scenario: 消除直接 DB 操作
- **GIVEN** 原 `polling_handler.go` 中有 13 处 `h.db.Model()` 直接 DB 操作
- **WHEN** 拆分后的 `polling_carddata_handler.go` 代码 Review
- **THEN** 文件中不出现 `h.db.``db.Model()``db.Where()` 等直接 GORM 调用;所有 DB 操作通过 Store 接口(`iotCardStore.UpdateXxx()``packageStore.UpdateXxx()`
#### Scenario: 跨月流量边界检测(必须保留)
- **GIVEN** cardID=100`current_month_start_date=2026-03-01`,当前日期为 `2026-04-02`
- **WHEN** Gateway 返回月度总流量 `500MB``processCard` 检测到跨月(当前月份首日 != current_month_start_date
- **THEN** 保存 `last_month_total_mb = 旧 current_month_total`;重置 `current_month_usage_mb = 500`(以 Gateway 新月份值为准);更新 `current_month_start_date = 2026-04-01`;记录流量历史到 `data_usage_records`
#### Scenario: 同月流量增量计算
- **GIVEN** cardID=200`current_month_start_date=2026-04-01``last_month_total_mb=100`,当前日期为 `2026-04-02`
- **WHEN** Gateway 返回月度总流量 `150MB`
- **THEN** 计算增量 `delta = 150 - 100 = 50MB`;更新 `current_month_usage_mb += 50`;更新 `last_month_total_mb = 150`
#### Scenario: Gateway 调用失败不丢失数据
- **GIVEN** cardID=300Gateway 返回网络超时
- **WHEN** `collectUsageData` 调用 Gateway 失败
- **THEN** 不更新 DB不写入错误数据记录 Warn 日志「流量查询失败: cardID=300, error=xxx」`requeueCard` 按较短间隔重试间隔重新入队Asynq 层面不触发重试Handler 返回 nilMaxRetry=0
---
### Requirement: polling_package_handler.go——套餐数据采集
**文件**`internal/task/polling_package_handler.go`< 200行
**构造函数依赖**
```go
func NewPollingPackageHandler(
base *PollingBase,
gateway GatewayClient,
packageStore PackageUsageStore,
stopResumeService StopResumeServiceInterface,
) *PollingPackageHandler
```
**方法列表**
- `Handle(ctx, task) error`Asynq 任务入口
- `processCard(ctx, cardID) error`:核心处理逻辑
**处理流程**
1. 获取并发控制
2. 调 Gateway 查询套餐信息(剩余流量、状态)
3. 写 Store更新 `tb_package_usage` 套餐状态/用量)
4.`stopResumeService.EvaluateAndAct(ctx, card, ...)`(停复机决策)
5. 重新入队,释放并发控制
#### Scenario: 套餐到期后触发复机判断
- **GIVEN** 卡因 `traffic_exhausted` 停机旧套餐已过期status=3Gateway 返回新套餐已激活
- **WHEN** `HandlePackageCheck` 执行,更新套餐状态后
- **THEN** 调 `EvaluateAndAct`StopResumeService 检测到新套餐有效、流量未耗尽触发复机Handler 不直接调用 resumeCard
#### Scenario: package Handler 不做停复机判断
- **GIVEN** `polling_package_handler.go` 代码 Review
- **WHEN** 检查文件内容
- **THEN** 文件中不出现 `shouldStopCard``hasAvailablePackage``stopCard``resumeCard` 等函数;套餐处理逻辑仅限于数据采集和 Store 写入
---
### Requirement: polling_protect_handler.go——保护期一致性检查
**文件**`internal/task/polling_protect_handler.go`< 200行
> **业务语义说明**:本 Handler 的核心目的是"确保保护期内卡状态与保护期方向一致,防止状态漂移"。
> 保护期**内**:强制修正状态(直接调 Gateway不走 EvaluateAndAct 三条件判断)。
> 保护期**结束后**:调 EvaluateAndAct 重新评估(按正常停复机逻辑处理)。
**构造函数依赖**
```go
func NewPollingProtectHandler(
base *PollingBase,
gateway GatewayClient, // 保护期内强制停复机需要直接调 Gateway
iotCardStore IotCardStore,
stopResumeService StopResumeServiceInterface,
) *PollingProtectHandler
```
**方法列表**
- `Handle(ctx, task) error`Asynq 任务入口
- `processCard(ctx, cardID) error`:核心处理逻辑
**处理流程**
1. 获取并发控制(并发满时 **requeue 后返回**,不丢弃)
2. 查 Store 获取卡信息
3. 前置跳过检查:
- 卡未实名(`real_name_status=0`)→ requeue直接返回
- 卡未绑定设备(`is_standalone=true`)→ requeue直接返回
4. 读取设备保护期 Redis Key`stop` 保护期 Key + `start` 保护期 Key
5. 根据保护期状态执行对应逻辑:
- **有 stop 保护期 + 卡在线network_status=1**:直接调 `gateway.StopCard`(强制修正),写 DB `stop_reason='protect'`
- **有 start 保护期 + 卡停机network_status=0**:直接调 `gateway.StartCard`(强制修正),清空 `stop_reason`
- **有保护期 + 状态已一致**:跳过(不调 Gateway不调 EvaluateAndAct
- **无保护期(保护期已结束)**:调 `stopResumeService.EvaluateAndAct(ctx, card)` 重新评估
6.`base.requeueCard` 按间隔重新入队
7. 释放并发控制
#### Scenario: stop 保护期内卡在线——强制停机
- **GIVEN** 卡已实名且绑定设备,设备有 stop 保护期(`polling:protect:stop:{deviceID}` Key 存在),但卡当前 `network_status=1`(在线,与保护期方向不一致)
- **WHEN** `HandleProtectConsistencyCheck` 执行
- **THEN** 检测到 stop 保护期存在且状态不一致;直接调 `gateway.StopCard`(不走 EvaluateAndAct强制修正DB 更新 `network_status=0, stop_reason='protect'`;记录 Info 日志「保护期强制停机: cardID=xxx, deviceID=xxx」
#### Scenario: start 保护期内卡停机——强制复机
- **GIVEN** 卡已实名且绑定设备,设备有 start 保护期(`polling:protect:start:{deviceID}` Key 存在),但卡当前 `network_status=0`(停机,与保护期方向不一致)
- **WHEN** `HandleProtectConsistencyCheck` 执行
- **THEN** 检测到 start 保护期存在且状态不一致;直接调 `gateway.StartCard`(不走 EvaluateAndAct强制修正DB 更新 `network_status=1, stop_reason=''`;记录 Info 日志「保护期强制复机: cardID=xxx, deviceID=xxx」
#### Scenario: 保护期内状态已一致——跳过
- **GIVEN** 设备有 stop 保护期,卡已是停机状态(`network_status=0`
- **WHEN** `HandleProtectConsistencyCheck` 执行
- **THEN** 检测到保护期存在且状态已一致;跳过,不调 Gateway不调 EvaluateAndAct直接 requeue
#### Scenario: 保护期已结束——调 EvaluateAndAct 重新评估
- **GIVEN** 卡在保护期内已停机(`stop_reason='protect'`),保护期 Key 已过期TTL = 0
- **WHEN** `HandleProtectConsistencyCheck` 执行,读取保护期 Key 不存在
- **THEN** 调 `stopResumeService.EvaluateAndAct(ctx, card)` 重新评估;若有有效套餐且已实名,触发复机;否则保持停机(`stop_reason` 更新为实际原因)
#### Scenario: 未实名卡跳过保护期逻辑
- **GIVEN** 卡 `real_name_status=0`(未实名)
- **WHEN** `HandleProtectConsistencyCheck` 执行
- **THEN** 检测到未实名,直接 requeue不检查保护期不调 Gateway
#### Scenario: 独立卡(未绑定设备)跳过
- **GIVEN** 卡 `is_standalone=true`(未绑定设备)
- **WHEN** `HandleProtectConsistencyCheck` 执行
- **THEN** 检测到独立卡,直接 requeue设备保护期与独立卡无关
---
### Requirement: PollingBase 支持 verboseLog 标志
`PollingBase` 结构体 SHALL 新增 `verboseLog bool` 字段,`NewPollingBase` 函数 SHALL 接受 `verboseLog bool` 参数并赋值。`cmd/worker/main.go` SHALL 从 `cfg.Polling.VerboseLog` 读取并传入。
#### Scenario: verboseLog 为 true 时详细日志启用
- **WHEN** `cfg.Polling.VerboseLog = true` 且 Worker 启动
- **THEN** `PollingBase.verboseLog = true`,所有 handler 在成功查询上游时输出 Info 日志
#### Scenario: verboseLog 为 false 时详细日志不输出
- **WHEN** `cfg.Polling.VerboseLog = false`(默认)
- **THEN** 成功查询上游时不输出额外日志,行为与修改前相同
---
### Requirement: 配置文件包含 polling.verbose_log 字段
`pkg/config/config.go` SHALL 新增 `PollingConfig struct``VerboseLog bool`,并加入 `Config` 结构体。`pkg/config/defaults/config.yaml` SHALL 包含 `polling.verbose_log: false`
#### Scenario: 默认配置 verbose_log 为 false
- **WHEN** 未设置 `JUNHONG_POLLING_VERBOSE_LOG` 环境变量
- **THEN** `cfg.Polling.VerboseLog = false`
#### Scenario: 环境变量覆盖开启 verbose_log
- **WHEN** 设置 `JUNHONG_POLLING_VERBOSE_LOG=true`
- **THEN** `cfg.Polling.VerboseLog = true`
---
### Requirement: realname handler 成功查询后输出 verbose 日志
`PollingRealnameHandler.Handle` SHALL 在 `gateway.QueryRealnameStatus` 成功返回后,当 `h.base.verboseLog` 为 true 时,以 `Info` 级别记录日志,包含:`card_id``iccid``real_status`bool`new_status`int`changed`bool
#### Scenario: verboseLog 开启时实名查询成功输出日志
- **WHEN** `verboseLog = true``QueryRealnameStatus` 成功返回
- **THEN** 输出包含 iccid + real_status + new_status + changed 的 Info 日志
#### Scenario: verboseLog 关闭时实名查询成功不输出额外日志
- **WHEN** `verboseLog = false``QueryRealnameStatus` 成功返回(状态未变)
- **THEN** 不输出 verbose 日志
---
### Requirement: carddata handler 成功查询后输出 verbose 日志
`PollingCarddataHandler.Handle` SHALL 在 `gateway.QueryFlow` 成功返回后,当 `h.base.verboseLog` 为 true 时,以 `Info` 级别记录日志,包含:`card_id``iccid``gateway_flow_mb`float64`increment_mb`float64`is_cross_month`bool
#### Scenario: verboseLog 开启时流量查询成功输出日志
- **WHEN** `verboseLog = true``QueryFlow` 成功返回
- **THEN** 输出包含 iccid + gateway_flow_mb + increment_mb + is_cross_month 的 Info 日志
#### Scenario: verboseLog 关闭时流量查询成功不输出额外日志
- **WHEN** `verboseLog = false``QueryFlow` 成功返回(流量无增量)
- **THEN** 不输出 verbose 日志
---
### Requirement: package handler 成功加载卡信息后输出 verbose 日志
`PollingPackageHandler.Handle` SHALL 在 `GetByID` 成功返回 `freshCard` 后、调用 `EvaluateAndAct` 之前,当 `h.base.verboseLog` 为 true 时,以 `Info` 级别记录日志,包含:`card_id``iccid``network_status`int`stop_reason`string
> package handler 不查询 Gatewayverbose log 记录的是本次评估时读取到的本地卡状态,便于排查"停复机评估为何结果不符预期"类问题。
#### Scenario: verboseLog 开启时套餐检查加载卡成功输出日志
- **WHEN** `verboseLog = true``iotCardStore.GetByID` 成功返回 `freshCard`
- **THEN** 输出包含 iccid + network_status + stop_reason 的 Info 日志
#### Scenario: verboseLog 关闭时套餐检查不输出额外日志
- **WHEN** `verboseLog = false`
- **THEN** 不输出 verbose 日志(现有 "套餐检查:执行停复机评估" Info 日志保持不变)
---
### Requirement: protect handler 成功执行保护期检查后输出 verbose 日志
`PollingProtectHandler.Handle` SHALL 在 switch 块执行完毕、调用 `UpdateFields(last_protect_check_at)` 之前,当 `h.base.verboseLog` 为 true 时,以 `Info` 级别记录日志,包含:`card_id``iccid``stop_protect`bool`start_protect`bool`network_status_at_check`int`action_taken`string
> 注:该日志在 switch 的所有分支强制停机、强制复机、EvaluateAndAct、无操作均成功执行后输出若某分支因 Gateway 失败提前 return则不输出已有 Error 日志覆盖)。`network_status_at_check` 反映检查时的卡状态(触发动作的原因),`action_taken` 反映实际执行了什么forced_stop / forced_resume / evaluate / no_action
#### Scenario: verboseLog 开启时保护期检查成功输出日志
- **WHEN** `verboseLog = true` 且 switch 块各分支均执行成功(未提前 return
- **THEN** 输出包含 iccid + stop_protect + start_protect + network_status_at_check + action_taken 的 Info 日志
#### Scenario: verboseLog 关闭时保护期检查不输出额外日志
- **WHEN** `verboseLog = false`
- **THEN** 不输出 verbose 日志