app-api/internal/task/device_persist.go

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
}