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) } } total := len(devices) persisted := len(succeeded) failed := total - persisted // 批量删除成功刷入的记录,失败的保留在 Redis 等待下次重试 if persisted > 0 { if err := t.rdb.HDel(ctx, "devices", succeeded...).Err(); err != nil { t.log.Error("redis HDel failed after persist", zap.Int("persisted", persisted), zap.Error(err), ) } else { t.log.Info("devices persisted to database and removed from redis", zap.Int("persisted", persisted), zap.Int("total", total), zap.Int("failed", failed), ) } } if failed > 0 { t.log.Warn("some devices failed to persist, kept in redis for retry", zap.Int("failed", failed), zap.Int("total", total), ) } } // 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 }