Files
junhong_cmp_fiber/internal/model/device.go
huang 434a8b0349
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m46s
feat: 轮询系统重构(分片队列 + 停复机统一 + Handler 拆分)
【核心变更】

1. 停复机逻辑统一(StopResumeService)
   - 新增 EvaluateAndAct 统一入口,封装三条件停复机判断
   - 停机条件:无套餐(no_package) / 流量耗尽(traffic_exhausted) / 未实名(not_realname)
   - 复机条件:stop_reason 合规 + 有套餐且未耗尽 + 已实名或行业卡
   - 修复设备套餐 Bug:hasValidPackage 按 device_id 查套餐,而非仅 iot_card_id
   - 设备维度停复机加幂等锁(Redis SetNX,TTL 30s),防止多卡并发重复调 Gateway

2. Redis 分片队列(PollingQueueManager)
   - 新建 queue_manager.go,封装所有轮询 Redis 操作
   - 16 分片 Sorted Set,Key 格式:polling:shard:{shardID}:queue:{taskType}
   - Lua 脚本原子出队(ZRANGEBYSCORE + 分批 ZREM),消除竞态窗口
   - 新增背压检测:队列深度超 50 万时 Scheduler 跳过该分片
   - RemoveFromAllQueues 覆盖 4 种任务类型(含 protect)

3. Handler 拆分(polling_handler.go 1360行 → 5个专注文件)
   - polling_base.go:共享基类(并发控制/卡缓存/重入队)
   - polling_realname_handler.go:实名采集,实名 0→1 时立即触发复机
   - polling_carddata_handler.go:流量采集,保留跨月边界检测逻辑
   - polling_package_handler.go:套餐采集,委托 EvaluateAndAct 决策
   - polling_protect_handler.go:保护期一致性检查,保护期内强制修正

4. 配置管理(PollingConfigManager)
   - 新建 config_manager.go,从 scheduler.go 提取配置职责
   - 内存缓存 + 5 分钟定时刷新,刷新失败保留原缓存
   - 修复 getCardCondition:停机卡返回 suspended,不再错配 activated 配置

5. 渐进式初始化(CardInitializer)
   - 新建 initializer.go,分批加载(每批 10 万),批次间 sleep 500ms
   - 过滤 enable_polling=false 的卡,初始化完成前 Scheduler 不出队

6. 卡生命周期服务(PollingLifecycleService)
   - 新建 lifecycle_service.go,替代已删除的 callbacks.go 和 api_callback.go
   - OnCardCreated/OnCardEnabled/OnCardStatusChanged 入队前检查 enable_polling

7. Scheduler 精简(1000+行 → 227行)
   - 保留纯调度循环:scheduleLoop + processShardSchedule + enqueueBatch
   - 保留每 10 秒触发套餐过期检测和流量重置
   - 移除所有 DB 操作、配置加载、卡初始化逻辑

8. 轮询管控 API(enable_polling)
   - 新增 PUT /api/admin/assets/:id/polling-status 接口
   - 支持对设备/卡维度开关轮询,关闭后从所有分片队列移除

9. 数据库迁移
   - 000103:tb_device 新增 enable_polling 字段(boolean, NOT NULL, DEFAULT true)
   - 000104:新增 suspended 轮询配置,为 activated 配置补全 protect_check_interval

【文件统计】
- 新增:19 个文件(handler × 5、polling 组件 × 4、迁移 × 3 等)
- 修改:20 个文件(bootstrap 注入、store 接口、monitoring 适配分片等)
- 删除:3 个文件(polling_handler.go、callbacks.go、api_callback.go)

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

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

152 lines
7.4 KiB
Go
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.
package model
import (
"encoding/json"
"strconv"
"time"
"gorm.io/gorm"
)
// Device 设备模型
// 物联网设备(如 GPS 追踪器、智能传感器)
// 通过 shop_id 区分所有权NULL=平台库存,有值=店铺所有
// 标识符说明virtual_no 为虚拟号/别名imei/sn 为设备真实标识
type Device struct {
gorm.Model
BaseModel `gorm:"embedded"`
VirtualNo string `gorm:"column:virtual_no;type:varchar(100);uniqueIndex:idx_device_virtual_no,where:deleted_at IS NULL;not null;comment:设备虚拟号/别名(用户友好的短标识)" json:"virtual_no"`
IMEI string `gorm:"column:imei;type:varchar(20);comment:设备IMEI(有蜂窝网络的设备标识,用于Gateway API调用)" json:"imei"`
SN string `gorm:"column:sn;type:varchar(100);comment:设备序列号(厂商唯一标识,预留字段)" json:"sn"`
DeviceName string `gorm:"column:device_name;type:varchar(255);comment:设备名称" json:"device_name"`
DeviceModel string `gorm:"column:device_model;type:varchar(100);comment:设备型号" json:"device_model"`
DeviceType string `gorm:"column:device_type;type:varchar(50);comment:设备类型" json:"device_type"`
MaxSimSlots int `gorm:"column:max_sim_slots;type:int;default:4;comment:最大插槽数量(默认4)" json:"max_sim_slots"`
Manufacturer string `gorm:"column:manufacturer;type:varchar(255);comment:制造商" json:"manufacturer"`
BatchNo string `gorm:"column:batch_no;type:varchar(100);comment:批次号" json:"batch_no"`
ShopID *uint `gorm:"column:shop_id;index;comment:店铺ID(NULL=平台库存,有值=店铺所有)" json:"shop_id,omitempty"`
Status int `gorm:"column:status;type:int;default:1;not null;comment:状态 1-在库 2-已分销 3-已激活 4-已停用" json:"status"`
ActivatedAt *time.Time `gorm:"column:activated_at;comment:激活时间" json:"activated_at"`
DeviceUsername string `gorm:"column:device_username;type:varchar(100);comment:设备登录用户名" json:"device_username"`
DevicePasswordEncrypted string `gorm:"column:device_password_encrypted;type:varchar(255);comment:设备登录密码(加密)" json:"device_password_encrypted"`
DeviceAPIEndpoint string `gorm:"column:device_api_endpoint;type:varchar(500);comment:设备API端点" json:"device_api_endpoint"`
SeriesID *uint `gorm:"column:series_id;index;comment:套餐系列ID(关联PackageSeries)" json:"series_id,omitempty"`
FirstCommissionPaid bool `gorm:"column:first_commission_paid;type:boolean;default:false;comment:一次性佣金是否已发放(废弃,使用按系列追踪)" json:"first_commission_paid"`
AccumulatedRecharge int64 `gorm:"column:accumulated_recharge;type:bigint;default:0;comment:累计充值金额(分,废弃,使用按系列追踪)" json:"accumulated_recharge"`
AccumulatedRechargeBySeriesJSON string `gorm:"column:accumulated_recharge_by_series;type:jsonb;default:'{}';comment:按套餐系列追踪的累计充值金额" json:"-"`
FirstRechargeTriggeredBySeriesJSON string `gorm:"column:first_recharge_triggered_by_series;type:jsonb;default:'{}';comment:按套餐系列追踪的首充触发状态" json:"-"`
AssetStatus int `gorm:"column:asset_status;type:int;not null;default:1;comment:业务状态 1-在库 2-已销售 3-已换货 4-已停用" json:"asset_status"`
Generation int `gorm:"column:generation;type:int;not null;default:1;comment:资产世代编号" json:"generation"`
// EnablePolling 是否参与轮询false 时设备下所有卡不加入轮询队列
EnablePolling bool `gorm:"column:enable_polling;not null;default:true" json:"enable_polling"`
// 以下字段由 Gateway sync-info 接口同步,有离线存储意义
OnlineStatus int `gorm:"column:online_status;type:int;not null;default:0;comment:在线状态0未知 1在线 2离线" json:"online_status"`
LastOnlineTime *time.Time `gorm:"column:last_online_time;comment:最后在线时间(来自 Gateway sync-info" json:"last_online_time,omitempty"`
SoftwareVersion string `gorm:"column:software_version;type:varchar(100);not null;default:'';comment:固件版本号" json:"software_version"`
SwitchMode string `gorm:"column:switch_mode;type:varchar(10);not null;default:'0';comment:切卡模式0=自动 1=手动" json:"switch_mode"`
LastGatewaySyncAt *time.Time `gorm:"column:last_gateway_sync_at;comment:最后一次 sync-info 同步时间" json:"last_gateway_sync_at,omitempty"`
}
// TableName 指定表名
func (Device) TableName() string {
return "tb_device"
}
func (d *Device) GetAccumulatedRechargeBySeriesMap() (map[uint]int64, error) {
result := make(map[uint]int64)
if d.AccumulatedRechargeBySeriesJSON == "" || d.AccumulatedRechargeBySeriesJSON == "{}" {
return result, nil
}
var raw map[string]int64
if err := json.Unmarshal([]byte(d.AccumulatedRechargeBySeriesJSON), &raw); err != nil {
return nil, err
}
for k, v := range raw {
id, err := strconv.ParseUint(k, 10, 64)
if err != nil {
continue
}
result[uint(id)] = v
}
return result, nil
}
func (d *Device) SetAccumulatedRechargeBySeriesMap(m map[uint]int64) error {
raw := make(map[string]int64)
for k, v := range m {
raw[strconv.FormatUint(uint64(k), 10)] = v
}
data, err := json.Marshal(raw)
if err != nil {
return err
}
d.AccumulatedRechargeBySeriesJSON = string(data)
return nil
}
func (d *Device) GetAccumulatedRechargeBySeries(seriesID uint) int64 {
m, err := d.GetAccumulatedRechargeBySeriesMap()
if err != nil {
return 0
}
return m[seriesID]
}
func (d *Device) AddAccumulatedRechargeBySeries(seriesID uint, amount int64) error {
m, err := d.GetAccumulatedRechargeBySeriesMap()
if err != nil {
m = make(map[uint]int64)
}
m[seriesID] += amount
return d.SetAccumulatedRechargeBySeriesMap(m)
}
func (d *Device) GetFirstRechargeTriggeredBySeriesMap() (map[uint]bool, error) {
result := make(map[uint]bool)
if d.FirstRechargeTriggeredBySeriesJSON == "" || d.FirstRechargeTriggeredBySeriesJSON == "{}" {
return result, nil
}
var raw map[string]bool
if err := json.Unmarshal([]byte(d.FirstRechargeTriggeredBySeriesJSON), &raw); err != nil {
return nil, err
}
for k, v := range raw {
id, err := strconv.ParseUint(k, 10, 64)
if err != nil {
continue
}
result[uint(id)] = v
}
return result, nil
}
func (d *Device) SetFirstRechargeTriggeredBySeriesMap(m map[uint]bool) error {
raw := make(map[string]bool)
for k, v := range m {
raw[strconv.FormatUint(uint64(k), 10)] = v
}
data, err := json.Marshal(raw)
if err != nil {
return err
}
d.FirstRechargeTriggeredBySeriesJSON = string(data)
return nil
}
func (d *Device) IsFirstRechargeTriggeredBySeries(seriesID uint) bool {
m, err := d.GetFirstRechargeTriggeredBySeriesMap()
if err != nil {
return false
}
return m[seriesID]
}
func (d *Device) SetFirstRechargeTriggeredBySeries(seriesID uint, triggered bool) error {
m, err := d.GetFirstRechargeTriggeredBySeriesMap()
if err != nil {
m = make(map[uint]bool)
}
m[seriesID] = triggered
return d.SetFirstRechargeTriggeredBySeriesMap(m)
}