junhong_cmp_fiber/openspec/specs/polling-task-handlers/spec.md

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 采集数据、写 Store（DB更新）、调 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 的卡信息已缓存在 Redis（TTL 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，将结果写入 Redis（TTL 5分钟）；返回卡信息

Scenario: 并发控制防止过载——并发满时必须 requeue

• GIVEN carddata 任务最大并发数配置为 50
• WHEN 同时有 60 个 carddata 任务尝试执行
• THEN 前 50 个获取到信号量正常执行；后 10 个 acquireConcurrency 返回 false；后 10 个任务调 requeueCard(ctx, cardID, taskType, time.Now()) 立即重入队，记录 Debug 日志「并发数已满，已重新入队: cardID=xxx」，返回 nil（Asynq 不报错，不重试）

⚠️ 正确性约束：Lua 脚本原子出队时卡已从 Redis Sorted Set 中删除。若并发满时直接 return nil 而不 requeue，该卡将永久消失（不再被轮询）。所有 Handler 必须在 acquireConcurrency 返回 false 时先调 requeueCard 再返回。

---

Requirement: polling_realname_handler.go——实名数据采集

文件：internal/task/polling_realname_handler.go（< 200行）

构造函数依赖（通过构造函数注入，禁止全局变量）：

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 时调用 EvaluateAndAct（S1 修复）

• 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行）

构造函数依赖：

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=300，Gateway 返回网络超时
• WHEN collectUsageData 调用 Gateway 失败
• THEN 不更新 DB（不写入错误数据）；记录 Warn 日志「流量查询失败: cardID=300, error=xxx」；调 requeueCard 按较短间隔（重试间隔）重新入队；Asynq 层面不触发重试（Handler 返回 nil，MaxRetry=0）

---

Requirement: polling_package_handler.go——套餐数据采集

文件：internal/task/polling_package_handler.go（< 200行）

构造函数依赖：

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=3），Gateway 返回新套餐已激活
• 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 重新评估（按正常停复机逻辑处理）。

构造函数依赖：

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（设备保护期与独立卡无关）