app-api/internal/task/impedance_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"
)

const redisHeadphoneImpedancesKey = "headphone_impedances"

type ImpedancePersistTask struct {
	rdb  *redis.Client
	repo *repository.HeadphoneImpedanceRepository
	log  *zap.Logger
}

func NewImpedancePersistTask(
	rdb *redis.Client,
	repo *repository.HeadphoneImpedanceRepository,
	log *zap.Logger,
) *ImpedancePersistTask {
	return &ImpedancePersistTask{rdb: rdb, repo: repo, log: log}
}

// Start 启动定时刷入协程，每 interval 执行一次
func (t *ImpedancePersistTask) Start(interval time.Duration) {
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()

		for range ticker.C {
			t.Persist()
		}
	}()
	t.log.Info("impedance persist task started", zap.String("interval", interval.String()))
}

// Persist 从 Redis Hash headphone_impedances 中读取数据刷入数据库
func (t *ImpedancePersistTask) Persist() {
	ctx := context.Background()

	records, err := t.rdb.HGetAll(ctx, redisHeadphoneImpedancesKey).Result()
	if err != nil {
		t.log.Error("redis HGetAll headphone_impedances failed", zap.Error(err))
		return
	}

	if len(records) == 0 {
		return
	}

	t.log.Info("persisting headphone impedances from redis", zap.Int("count", len(records)))

	var succeeded []string

	for field, jsonStr := range records {
		var info struct {
			MacAddr            string `json:"mac_addr"`
			DeviceModel        string `json:"device_model"`
			ImpedanceOhm       int    `json:"impedance_ohm"`
			HeadphoneBrand     string `json:"headphone_brand"`
			HeadphoneModel     string `json:"headphone_model"`
			HeadphoneBrandNorm string `json:"headphone_brand_norm"`
			HeadphoneModelNorm string `json:"headphone_model_norm"`
			IpAddr             string `json:"ip_addr"`
		}

		if err := json.Unmarshal([]byte(jsonStr), &info); err != nil {
			t.log.Error("unmarshal impedance info failed", zap.String("field", field), zap.Error(err))
			continue
		}

		if t.persistOne(ctx, info) {
			succeeded = append(succeeded, field)
		}
	}

	total := len(records)
	persisted := len(succeeded)
	failed := total - persisted

	if persisted > 0 {
		if err := t.rdb.HDel(ctx, redisHeadphoneImpedancesKey, succeeded...).Err(); err != nil {
			t.log.Error("redis HDel failed after persist",
				zap.Int("persisted", persisted),
				zap.Error(err),
			)
		} else {
			t.log.Info("headphone impedances 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 headphone impedances failed to persist, kept in redis for retry",
			zap.Int("failed", failed),
			zap.Int("total", total),
		)
	}
}

func (t *ImpedancePersistTask) persistOne(ctx context.Context, info struct {
	MacAddr            string `json:"mac_addr"`
	DeviceModel        string `json:"device_model"`
	ImpedanceOhm       int    `json:"impedance_ohm"`
	HeadphoneBrand     string `json:"headphone_brand"`
	HeadphoneModel     string `json:"headphone_model"`
	HeadphoneBrandNorm string `json:"headphone_brand_norm"`
	HeadphoneModelNorm string `json:"headphone_model_norm"`
	IpAddr             string `json:"ip_addr"`
}) bool {
	existing, err := t.repo.FindByMacAndNorm(ctx, info.MacAddr, info.HeadphoneBrandNorm, info.HeadphoneModelNorm)
	if err != nil {
		t.log.Error("find headphone impedance failed",
			zap.String("mac", info.MacAddr),
			zap.String("brand_norm", info.HeadphoneBrandNorm),
			zap.String("model_norm", info.HeadphoneModelNorm),
			zap.Error(err),
		)
		return false
	}

	rec := model.UserHeadphoneImpedance{
		MacAddr:            info.MacAddr,
		DeviceModel:        info.DeviceModel,
		ImpedanceOhm:       info.ImpedanceOhm,
		HeadphoneBrand:     info.HeadphoneBrand,
		HeadphoneModel:     info.HeadphoneModel,
		HeadphoneBrandNorm: info.HeadphoneBrandNorm,
		HeadphoneModelNorm: info.HeadphoneModelNorm,
		IpAddr:             info.IpAddr,
	}

	if existing == nil {
		if err := t.repo.Insert(ctx, rec); err != nil {
			t.log.Error("insert user_headphone_impedance failed",
				zap.String("mac", info.MacAddr),
				zap.Error(err),
			)
			return false
		}
	} else {
		rec.ID = existing.ID
		if err := t.repo.Update(ctx, rec); err != nil {
			t.log.Error("update user_headphone_impedance failed",
				zap.String("mac", info.MacAddr),
				zap.Error(err),
			)
			return false
		}
	}

	return true
}