feat(device): 增加设备信息持久化和OTA升级功能

- 新增设备持久化定时任务，每5分钟将Redis中采集的设备信息同步数据库 - 新增UserDevice和UserActive模型及相应数据库表和操作接口 - 设备信息上报中获取客户端公网IP，替代原X-Forwarded-For头部 - 新建OTA功能模块，实现OTA固件升级信息查询接口 - 支持OTA黑名单过滤与定向升级设备判断 - 设计OTA相关数据库结构：ota、black_list、ota_target_device表 - 缓存预热新增按品牌分组预热型号缓存 - 依赖注入新增OTA Repository及Handler路由配置 - 实现客户端公网IP提取逻辑，支持多种代理头部优先级识别
2026-05-28 18:26:20 +08:00
parent b2c11adeba
commit f3e3e82f52
16 changed files with 1078 additions and 9 deletions
@@ -0,0 +1,166 @@
+package task
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/luxsin/app-api/internal/model"
+	"github.com/luxsin/app-api/internal/repository"
+	"github.com/redis/go-redis/v9"
+	"go.uber.org/zap"
+)
+
+type DevicePersistTask struct {
+	rdb  *redis.Client
+	repo *repository.DeviceRepository
+	log  *zap.Logger
+}
+
+func NewDevicePersistTask(rdb *redis.Client, repo *repository.DeviceRepository, log *zap.Logger) *DevicePersistTask {
+	return &DevicePersistTask{rdb: rdb, repo: repo, log: log}
+}
+
+// Start 启动定时刷入协程，每 interval 执行一次
+func (t *DevicePersistTask) Start(interval time.Duration) {
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		for range ticker.C {
+			t.Persist()
+		}
+	}()
+	t.log.Info("device persist task started", zap.String("interval", interval.String()))
+}
+
+// Persist 从 Redis Hash devices 中读取数据刷入数据库
+func (t *DevicePersistTask) Persist() {
+	ctx := context.Background()
+
+	// 获取 Redis Hash 所有数据
+	devices, err := t.rdb.HGetAll(ctx, "devices").Result()
+	if err != nil {
+		t.log.Error("redis HGetAll devices failed", zap.Error(err))
+		return
+	}
+
+	if len(devices) == 0 {
+		return
+	}
+
+	t.log.Info("persisting devices from redis", zap.Int("count", len(devices)))
+
+	var succeeded []string
+
+	for mac, jsonStr := range devices {
+		var info struct {
+			MacAddr    string `json:"mac_addr"`
+			Model      string `json:"model"`
+			ActiveDate string `json:"active_date"`
+			IpAddr     string `json:"ip_addr"`
+			Ver        string `json:"ver"`
+		}
+
+		if err := json.Unmarshal([]byte(jsonStr), &info); err != nil {
+			t.log.Error("unmarshal device info failed", zap.String("mac", mac), zap.Error(err))
+			continue
+		}
+
+		// 修复旧的无效日期格式（之前 active_date 使用了 Java 格式 yyyy-MM-dd）
+		if info.ActiveDate == "" || info.ActiveDate == "yyyy-MM-dd" {
+			info.ActiveDate = time.Now().Format("2006-01-02")
+		}
+
+		deviceOk := t.persistDevice(ctx, info)
+		activeOk := t.persistActive(ctx, info)
+
+		if deviceOk && activeOk {
+			succeeded = append(succeeded, mac)
+		}
+	}
+
+	// 批量删除成功处理的记录
+	if len(succeeded) > 0 {
+		if err := t.rdb.HDel(ctx, "devices", succeeded...).Err(); err != nil {
+			t.log.Error("redis HDel failed", zap.Int("count", len(succeeded)), zap.Error(err))
+		} else {
+			t.log.Info("devices persisted and removed from redis", zap.Int("count", len(succeeded)))
+		}
+	}
+}
+
+// persistDevice 处理 user_device 表写入
+func (t *DevicePersistTask) persistDevice(ctx context.Context, info struct {
+	MacAddr    string `json:"mac_addr"`
+	Model      string `json:"model"`
+	ActiveDate string `json:"active_date"`
+	IpAddr     string `json:"ip_addr"`
+	Ver        string `json:"ver"`
+}) bool {
+	existing, err := t.repo.FindDeviceByMac(ctx, info.MacAddr)
+	if err != nil {
+		t.log.Error("find device by mac failed", zap.String("mac", info.MacAddr), zap.Error(err))
+		return false
+	}
+
+	if existing == nil {
+		d := model.UserDevice{
+			MacAddr: info.MacAddr,
+			Model:   info.Model,
+		}
+		if info.Ver != "" {
+			d.Ver = &info.Ver
+		}
+		if err := t.repo.InsertDevice(ctx, d); err != nil {
+			t.log.Error("insert user_device failed", zap.String("mac", info.MacAddr), zap.Error(err))
+			return false
+		}
+	} else {
+		var ver *string
+		if info.Ver != "" {
+			ver = &info.Ver
+		}
+		if err := t.repo.UpdateDeviceVer(ctx, existing.ID, ver); err != nil {
+			t.log.Error("update user_device failed", zap.String("mac", info.MacAddr), zap.Error(err))
+			return false
+		}
+	}
+
+	return true
+}
+
+// persistActive 处理 user_active 表写入
+func (t *DevicePersistTask) persistActive(ctx context.Context, info struct {
+	MacAddr    string `json:"mac_addr"`
+	Model      string `json:"model"`
+	ActiveDate string `json:"active_date"`
+	IpAddr     string `json:"ip_addr"`
+	Ver        string `json:"ver"`
+}) bool {
+	dbActive, err := t.repo.FindActiveByMacAndDate(ctx, info.MacAddr, info.ActiveDate)
+	if err != nil {
+		t.log.Error("find active by mac and date failed", zap.String("mac", info.MacAddr), zap.Error(err))
+		return false
+	}
+
+	if dbActive == nil {
+		a := model.UserActive{
+			MacAddr:    info.MacAddr,
+			Model:      info.Model,
+			ActiveDate: info.ActiveDate,
+			IpAddr:     info.IpAddr,
+		}
+		if err := t.repo.InsertActive(ctx, a); err != nil {
+			t.log.Error("insert user_active failed", zap.String("mac", info.MacAddr), zap.Error(err))
+			return false
+		}
+	} else {
+		if err := t.repo.UpdateActiveIp(ctx, dbActive.ID, info.IpAddr); err != nil {
+			t.log.Error("update user_active failed", zap.String("mac", info.MacAddr), zap.Error(err))
+			return false
+		}
+	}
+
+	return true
+}