package redis import ( "context" "encoding/json" "errors" "fmt" "fwh-toolbox-server/cache" "fwh-toolbox-server/domain" "os" "strconv" "sync" "time" rds "github.com/redis/go-redis/v9" "go.uber.org/zap" ) var ( emsparaCtx = context.Background() emsparaKeyPrefix = "fwhtoolbox:emspara:" emsparaVidKey = emsparaKeyPrefix + "VID" emsparaPidKey = emsparaKeyPrefix + "PID" ) // emsparaReloadInterval 字典缺失时两次自动重新加载的最小间隔,避免加载失败时每次查询都去读 SQLite const emsparaReloadInterval = time.Minute var ( emsparaLoader func() error // 由 utils/emspara.Init 注册:SQLite -> Redis emsparaReloadMu sync.Mutex emsparaLastReload time.Time ) // SetEmsparaLoader 注册字典加载函数。查询时发现 Redis 中没有字典(整个 key 不存在)会调用它重新加载。 // 由 utils/emspara 注册,避免 repositories 反向依赖 SQLite。 func SetEmsparaLoader(fn func() error) { emsparaReloadMu.Lock() defer emsparaReloadMu.Unlock() emsparaLoader = fn } // UpdateEmspara 用 SQLite 中的字典整体替换 Redis 中的字典。 // 先写入本进程专用的临时 key,再 RENAME 为正式 key:替换是原子的,读取方不会看到空字典或半份字典, // 多个进程(主服务、fwh-agent-mcp)同时加载也不会互相覆盖写到一半的数据。 func UpdateEmspara(vidMacros []domain.ParamsVidMacroModel, pidMacros []domain.ParamsPidMacroModel) error { if len(vidMacros) == 0 || len(pidMacros) == 0 { return fmt.Errorf("EMS 参数字典为空(VID %d 条,PID %d 条),不替换 Redis 中的数据", len(vidMacros), len(pidMacros)) } vid := make(map[string]any, len(vidMacros)) for _, m := range vidMacros { val, err := json.Marshal(map[string]string{"name": m.MacroName, "desc": m.Description}) if err != nil { return err } vid[strconv.FormatUint(uint64(m.ID), 10)] = val } pid := make(map[string]any, len(pidMacros)) for _, m := range pidMacros { val, err := json.Marshal(map[string]string{"name": m.MacroName, "desc": m.Description}) if err != nil { return err } pid[strconv.FormatUint(uint64(m.ID), 10)] = val } ctx, cancel := context.WithTimeout(emsparaCtx, 30*time.Second) defer cancel() suffix := fmt.Sprintf(":loading:%d:%d", os.Getpid(), time.Now().UnixNano()) _, err := cache.Pipeline(ctx, func(p rds.Pipeliner) error { for key, data := range map[string]map[string]any{emsparaVidKey: vid, emsparaPidKey: pid} { tmp := key + suffix p.HSet(ctx, tmp, data) // 写到一半失败时临时 key 自动过期;RENAME 会带上 TTL,替换后再 PERSIST p.Expire(ctx, tmp, 10*time.Minute) p.Rename(ctx, tmp, key) p.Persist(ctx, key) } return nil }) return err } // emsparaInfo 按编号查询字典;字典整个不存在时触发一次重新加载后重试 func emsparaInfo(key string, id uint) (string, string, error) { field := strconv.FormatUint(uint64(id), 10) v, err := emsparaGet(key, field) if errors.Is(err, rds.Nil) && emsparaReloadIfMissing(key) { v, err = emsparaGet(key, field) } if err != nil { return "", "", err } var m map[string]string if err := json.Unmarshal([]byte(v), &m); err != nil { return "", "", err } return m["name"], m["desc"], nil } func emsparaGet(key, field string) (string, error) { ctx, cancel := context.WithTimeout(emsparaCtx, 5*time.Second) defer cancel() return cache.HGet(ctx, key, field) } // emsparaReloadIfMissing 字典 key 不存在时重新加载;返回是否已重新加载成功。 // 只是单个编号不在字典中(key 存在)时不加载。 func emsparaReloadIfMissing(key string) bool { if exists, err := emsparaExists(key); err != nil || exists { return false } emsparaReloadMu.Lock() defer emsparaReloadMu.Unlock() // 等锁期间可能已被其他查询重新加载 if exists, err := emsparaExists(key); err == nil && exists { return true } if emsparaLoader == nil || time.Since(emsparaLastReload) < emsparaReloadInterval { return false } emsparaLastReload = time.Now() if err := emsparaLoader(); err != nil { zap.L().Error("[emspara] Redis 中没有 EMS 参数字典,重新加载失败", zap.String("key", key), zap.Error(err)) return false } zap.L().Info("[emspara] Redis 中没有 EMS 参数字典,已重新加载", zap.String("key", key)) return true } func emsparaExists(key string) (bool, error) { ctx, cancel := context.WithTimeout(emsparaCtx, 5*time.Second) defer cancel() return cache.Exists(ctx, key) } func GetEmsparaVidInfo(id uint) (string, string, error) { return emsparaInfo(emsparaVidKey, id) } func GetEmsparaPidInfo(id uint) (string, string, error) { return emsparaInfo(emsparaPidKey, id) }