emsparaRepo.go 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  1. package redis
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "fwh-toolbox-server/cache"
  8. "fwh-toolbox-server/domain"
  9. "os"
  10. "strconv"
  11. "sync"
  12. "time"
  13. rds "github.com/redis/go-redis/v9"
  14. "go.uber.org/zap"
  15. )
  16. var (
  17. emsparaCtx = context.Background()
  18. emsparaKeyPrefix = "fwhtoolbox:emspara:"
  19. emsparaVidKey = emsparaKeyPrefix + "VID"
  20. emsparaPidKey = emsparaKeyPrefix + "PID"
  21. )
  22. // emsparaReloadInterval 字典缺失时两次自动重新加载的最小间隔,避免加载失败时每次查询都去读 SQLite
  23. const emsparaReloadInterval = time.Minute
  24. var (
  25. emsparaLoader func() error // 由 utils/emspara.Init 注册:SQLite -> Redis
  26. emsparaReloadMu sync.Mutex
  27. emsparaLastReload time.Time
  28. )
  29. // SetEmsparaLoader 注册字典加载函数。查询时发现 Redis 中没有字典(整个 key 不存在)会调用它重新加载。
  30. // 由 utils/emspara 注册,避免 repositories 反向依赖 SQLite。
  31. func SetEmsparaLoader(fn func() error) {
  32. emsparaReloadMu.Lock()
  33. defer emsparaReloadMu.Unlock()
  34. emsparaLoader = fn
  35. }
  36. // UpdateEmspara 用 SQLite 中的字典整体替换 Redis 中的字典。
  37. // 先写入本进程专用的临时 key,再 RENAME 为正式 key:替换是原子的,读取方不会看到空字典或半份字典,
  38. // 多个进程(主服务、fwh-agent-mcp)同时加载也不会互相覆盖写到一半的数据。
  39. func UpdateEmspara(vidMacros []domain.ParamsVidMacroModel, pidMacros []domain.ParamsPidMacroModel) error {
  40. if len(vidMacros) == 0 || len(pidMacros) == 0 {
  41. return fmt.Errorf("EMS 参数字典为空(VID %d 条,PID %d 条),不替换 Redis 中的数据", len(vidMacros), len(pidMacros))
  42. }
  43. vid := make(map[string]any, len(vidMacros))
  44. for _, m := range vidMacros {
  45. val, err := json.Marshal(map[string]string{"name": m.MacroName, "desc": m.Description})
  46. if err != nil {
  47. return err
  48. }
  49. vid[strconv.FormatUint(uint64(m.ID), 10)] = val
  50. }
  51. pid := make(map[string]any, len(pidMacros))
  52. for _, m := range pidMacros {
  53. val, err := json.Marshal(map[string]string{"name": m.MacroName, "desc": m.Description})
  54. if err != nil {
  55. return err
  56. }
  57. pid[strconv.FormatUint(uint64(m.ID), 10)] = val
  58. }
  59. ctx, cancel := context.WithTimeout(emsparaCtx, 30*time.Second)
  60. defer cancel()
  61. suffix := fmt.Sprintf(":loading:%d:%d", os.Getpid(), time.Now().UnixNano())
  62. _, err := cache.Pipeline(ctx, func(p rds.Pipeliner) error {
  63. for key, data := range map[string]map[string]any{emsparaVidKey: vid, emsparaPidKey: pid} {
  64. tmp := key + suffix
  65. p.HSet(ctx, tmp, data)
  66. // 写到一半失败时临时 key 自动过期;RENAME 会带上 TTL,替换后再 PERSIST
  67. p.Expire(ctx, tmp, 10*time.Minute)
  68. p.Rename(ctx, tmp, key)
  69. p.Persist(ctx, key)
  70. }
  71. return nil
  72. })
  73. return err
  74. }
  75. // emsparaInfo 按编号查询字典;字典整个不存在时触发一次重新加载后重试
  76. func emsparaInfo(key string, id uint) (string, string, error) {
  77. field := strconv.FormatUint(uint64(id), 10)
  78. v, err := emsparaGet(key, field)
  79. if errors.Is(err, rds.Nil) && emsparaReloadIfMissing(key) {
  80. v, err = emsparaGet(key, field)
  81. }
  82. if err != nil {
  83. return "", "", err
  84. }
  85. var m map[string]string
  86. if err := json.Unmarshal([]byte(v), &m); err != nil {
  87. return "", "", err
  88. }
  89. return m["name"], m["desc"], nil
  90. }
  91. func emsparaGet(key, field string) (string, error) {
  92. ctx, cancel := context.WithTimeout(emsparaCtx, 5*time.Second)
  93. defer cancel()
  94. return cache.HGet(ctx, key, field)
  95. }
  96. // emsparaReloadIfMissing 字典 key 不存在时重新加载;返回是否已重新加载成功。
  97. // 只是单个编号不在字典中(key 存在)时不加载。
  98. func emsparaReloadIfMissing(key string) bool {
  99. if exists, err := emsparaExists(key); err != nil || exists {
  100. return false
  101. }
  102. emsparaReloadMu.Lock()
  103. defer emsparaReloadMu.Unlock()
  104. // 等锁期间可能已被其他查询重新加载
  105. if exists, err := emsparaExists(key); err == nil && exists {
  106. return true
  107. }
  108. if emsparaLoader == nil || time.Since(emsparaLastReload) < emsparaReloadInterval {
  109. return false
  110. }
  111. emsparaLastReload = time.Now()
  112. if err := emsparaLoader(); err != nil {
  113. zap.L().Error("[emspara] Redis 中没有 EMS 参数字典,重新加载失败", zap.String("key", key), zap.Error(err))
  114. return false
  115. }
  116. zap.L().Info("[emspara] Redis 中没有 EMS 参数字典,已重新加载", zap.String("key", key))
  117. return true
  118. }
  119. func emsparaExists(key string) (bool, error) {
  120. ctx, cancel := context.WithTimeout(emsparaCtx, 5*time.Second)
  121. defer cancel()
  122. return cache.Exists(ctx, key)
  123. }
  124. func GetEmsparaVidInfo(id uint) (string, string, error) {
  125. return emsparaInfo(emsparaVidKey, id)
  126. }
  127. func GetEmsparaPidInfo(id uint) (string, string, error) {
  128. return emsparaInfo(emsparaPidKey, id)
  129. }