package service import ( "cmp" "context" "errors" "math" "slices" "sync" "time" "with_you/internal/cache" "with_you/internal/config" "with_you/internal/repository" "github.com/redis/go-redis/v9" "go.uber.org/zap" ) type hotRankScored struct { idx int norm float64 } // HotRankWorker 定时在候选池上重算热门分,写入 Redis ZSET(仅 TopN)及 top_ids(Redis+本地缓存) type HotRankWorker struct { cfg *config.Config postRepo repository.PostRepository channelRepo repository.ChannelRepository c cache.Cache mu sync.Mutex stopOnce sync.Once stopCh chan struct{} doneCh chan struct{} } func NewHotRankWorker(cfg *config.Config, postRepo repository.PostRepository, channelRepo repository.ChannelRepository, c cache.Cache) *HotRankWorker { return &HotRankWorker{ cfg: cfg, postRepo: postRepo, channelRepo: channelRepo, c: c, stopCh: make(chan struct{}), doneCh: make(chan struct{}), } } // Start 启动定时任务(启动时立即执行一次) func (w *HotRankWorker) Start(ctx context.Context) { if w == nil || w.cfg == nil || !w.cfg.HotRank.Enabled || w.c == nil { close(w.doneCh) return } interval := time.Duration(w.cfg.HotRank.RefreshIntervalSeconds) * time.Second if interval < 30*time.Second { interval = 30 * time.Second } go func() { defer close(w.doneCh) if err := w.Refresh(ctx); err != nil { if !errors.Is(err, cache.ErrKeyNotFound) { zap.L().Warn("hot rank initial refresh failed", zap.Error(err)) } } ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-w.stopCh: return case <-ticker.C: if err := w.Refresh(ctx); err != nil { if !errors.Is(err, cache.ErrKeyNotFound) { zap.L().Warn("hot rank refresh failed", zap.Error(err)) } } } } }() } // Stop 停止 worker func (w *HotRankWorker) Stop() { if w == nil { return } w.stopOnce.Do(func() { close(w.stopCh) }) <-w.doneCh } // Refresh 立即重算热门分(线程安全) // 算法:候选 ID = 上一期 TopN(Redis)∪ 时间窗内新发帖子 ID,截断至 CandidateCap; // 仅对候选拉取统计字段并打分,min-max 归一化;已发布置顶帖固定排在最前,其后接热门至 TopN; // 写入 ZSET(置顶分数高于热门)及 top_ids。 func (w *HotRankWorker) Refresh(ctx context.Context) error { if w == nil || w.c == nil { return cache.ErrKeyNotFound } w.mu.Lock() defer w.mu.Unlock() topN := w.cfg.HotRank.TopN if topN <= 0 { topN = 100 } topN = min(topN, 500) recentHours := w.cfg.HotRank.RecentWindowHours if recentHours <= 0 { recentHours = 168 } recentLimit := w.cfg.HotRank.RecentFetchLimit if recentLimit <= 0 { recentLimit = 500 } candidateCap := w.cfg.HotRank.CandidateCap if candidateCap <= 0 { candidateCap = 800 } zsetKey := cache.HotRankZSetKey() topIDsKey := cache.HotRankTopIDsKey() pinned, err := w.postRepo.ListPinnedPublishedPostIDs() if err != nil { return err } pinned = dedupeIDsPreserveOrder(pinned) incumbent, _ := w.c.ZRevRange(ctx, zsetKey, 0, int64(topN-1)) since := time.Now().Add(-time.Duration(recentHours) * time.Hour) recentIDs, err := w.postRepo.ListPublishedPostIDsSince(since, recentLimit) if err != nil { return err } candidateIDs := mergeHotRankCandidates(incumbent, recentIDs, candidateCap) if len(candidateIDs) == 0 && len(pinned) == 0 { if err := w.c.ZReplaceSortedSet(ctx, zsetKey, nil); err != nil { return err } w.c.Delete(topIDsKey) return nil } var snapshots []repository.PostHotSnapshot var ranked []hotRankScored if len(candidateIDs) > 0 { snapshots, err = w.postRepo.GetPostHotSnapshotsByIDs(candidateIDs) if err != nil { return err } } if len(snapshots) > 0 { now := time.Now() raw := make([]float64, len(snapshots)) for i := range snapshots { p := &snapshots[i] base := hotRankBaseScore(p.LikesCount, p.CommentsCount, p.FavoritesCount, p.SharesCount, p.ViewsCount) tHours := now.Sub(p.CreatedAt).Hours() if tHours < 0 { tHours = 0 } denom := math.Pow(tHours+2.0, 0.5) if denom < 1e-9 { denom = 1e-9 } raw[i] = base / denom } minR := slices.Min(raw) maxR := slices.Max(raw) ranked = make([]hotRankScored, len(snapshots)) for i := range snapshots { var norm float64 if maxR > minR { norm = (raw[i] - minR) / (maxR - minR) } else { norm = 1 } ranked[i] = hotRankScored{idx: i, norm: norm} } slices.SortFunc(ranked, func(a, b hotRankScored) int { if a.norm != b.norm { return cmp.Compare(b.norm, a.norm) // 降序 } if snapshots[a.idx].CreatedAt.After(snapshots[b.idx].CreatedAt) { return -1 } else if snapshots[a.idx].CreatedAt.Before(snapshots[b.idx].CreatedAt) { return 1 } return 0 }) } pinnedSet := make(map[string]struct{}, len(pinned)) for _, id := range pinned { pinnedSet[id] = struct{}{} } hotSlots := topN - len(pinned) if hotSlots < 0 { pinned = pinned[:topN] pinnedSet = make(map[string]struct{}, len(pinned)) for _, id := range pinned { pinnedSet[id] = struct{}{} } hotSlots = 0 } type hotPick struct { id string norm float64 } hotPicks := make([]hotPick, 0, hotSlots) for _, r := range ranked { id := snapshots[r.idx].ID if _, isPin := pinnedSet[id]; isPin { continue } hotPicks = append(hotPicks, hotPick{id: id, norm: r.norm}) if len(hotPicks) >= hotSlots { break } } orderedIDs := make([]string, 0, len(pinned)+len(hotPicks)) orderedIDs = append(orderedIDs, pinned...) for _, h := range hotPicks { orderedIDs = append(orderedIDs, h.id) } const pinnedScoreBase = 1000.0 zMembers := make([]redis.Z, 0, len(orderedIDs)) for i, id := range pinned { zMembers = append(zMembers, redis.Z{ Score: pinnedScoreBase - float64(i)*1e-6, Member: id, }) } for _, h := range hotPicks { zMembers = append(zMembers, redis.Z{Score: h.norm * 0.99, Member: h.id}) } if err := w.c.ZReplaceSortedSet(ctx, zsetKey, zMembers); err != nil { return err } ttl := time.Duration(w.cfg.HotRank.RefreshIntervalSeconds) * time.Second * 4 if ttl < 10*time.Minute { ttl = 10 * time.Minute } w.c.Set(topIDsKey, orderedIDs, ttl) zap.L().Info("hot rank refreshed", zap.Int("pinned", len(pinned)), zap.Int("candidates", len(candidateIDs)), zap.Int("scored", len(snapshots)), zap.Int("top_n", len(zMembers)), ) if err := w.refreshChannelRanks(ctx, ttl); err != nil { zap.L().Warn("hot rank channel refresh failed", zap.Error(err)) } return nil } func dedupeIDsPreserveOrder(ids []string) []string { if len(ids) <= 1 { return ids } seen := make(map[string]struct{}, len(ids)) out := make([]string, 0, len(ids)) for _, id := range ids { if id == "" { continue } if _, ok := seen[id]; ok { continue } seen[id] = struct{}{} out = append(out, id) } return out } func mergeHotRankCandidates(incumbent, recent []string, cap int) []string { if cap <= 0 { return nil } seen := make(map[string]struct{}, cap) out := make([]string, 0, cap) for _, id := range incumbent { if id == "" { continue } if _, ok := seen[id]; ok { continue } seen[id] = struct{}{} out = append(out, id) if len(out) >= cap { return out } } for _, id := range recent { if id == "" { continue } if _, ok := seen[id]; ok { continue } seen[id] = struct{}{} out = append(out, id) if len(out) >= cap { break } } return out } func hotRankBaseScore(likes, comments, favorites, shares, views int) float64 { return float64(likes)*3 + float64(comments)*5 + float64(favorites)*4 + float64(shares)*6 + float64(views)*0.15 } func (w *HotRankWorker) refreshChannelRanks( ctx context.Context, ttl time.Duration, ) error { if w.channelRepo == nil { return nil } channels, err := w.channelRepo.ListActive() if err != nil { return err } if len(channels) == 0 { return nil } topN := w.cfg.HotRank.TopN if topN <= 0 { topN = 100 } topN = min(topN, 500) for _, ch := range channels { chCandidateIDs, cerr := w.postRepo.ListPublishedPostIDsByChannel(ch.ID, w.cfg.HotRank.RecentWindowHours, w.cfg.HotRank.CandidateCap) if cerr != nil || len(chCandidateIDs) == 0 { continue } var chRanked []hotRankScored chSnapshots, ferr := w.postRepo.GetPostHotSnapshotsByIDs(chCandidateIDs) if ferr != nil || len(chSnapshots) == 0 { continue } now := time.Now() raw := make([]float64, len(chSnapshots)) for i := range chSnapshots { p := &chSnapshots[i] base := hotRankBaseScore(p.LikesCount, p.CommentsCount, p.FavoritesCount, p.SharesCount, p.ViewsCount) tHours := now.Sub(p.CreatedAt).Hours() if tHours < 0 { tHours = 0 } denom := math.Pow(tHours+2.0, 0.5) if denom < 1e-9 { denom = 1e-9 } raw[i] = base / denom } minR := slices.Min(raw) maxR := slices.Max(raw) chRanked = make([]hotRankScored, len(chSnapshots)) for i := range chSnapshots { var norm float64 if maxR > minR { norm = (raw[i] - minR) / (maxR - minR) } else { norm = 1 } chRanked[i] = hotRankScored{idx: i, norm: norm} } slices.SortFunc(chRanked, func(a, b hotRankScored) int { if a.norm != b.norm { return cmp.Compare(b.norm, a.norm) } if chSnapshots[a.idx].CreatedAt.After(chSnapshots[b.idx].CreatedAt) { return -1 } else if chSnapshots[a.idx].CreatedAt.Before(chSnapshots[b.idx].CreatedAt) { return 1 } return 0 }) chPinned, _ := w.postRepo.ListPinnedPublishedPostIDsByChannel(ch.ID) chPinned = dedupeIDsPreserveOrder(chPinned) chPinnedSet := make(map[string]struct{}, len(chPinned)) for _, id := range chPinned { chPinnedSet[id] = struct{}{} } hotSlots := topN - len(chPinned) if hotSlots < 0 { chPinned = chPinned[:topN] hotSlots = 0 } type hotPick struct { id string norm float64 } hotPicks := make([]hotPick, 0, hotSlots) for _, r := range chRanked { id := chSnapshots[r.idx].ID if _, isPin := chPinnedSet[id]; isPin { continue } hotPicks = append(hotPicks, hotPick{id: id, norm: r.norm}) if len(hotPicks) >= hotSlots { break } } chOrderedIDs := make([]string, 0, len(chPinned)+len(hotPicks)) chOrderedIDs = append(chOrderedIDs, chPinned...) for _, h := range hotPicks { chOrderedIDs = append(chOrderedIDs, h.id) } const pinnedScoreBase = 1000.0 chZMembers := make([]redis.Z, 0, len(chOrderedIDs)) for i, id := range chPinned { chZMembers = append(chZMembers, redis.Z{ Score: pinnedScoreBase - float64(i)*1e-6, Member: id, }) } for _, h := range hotPicks { chZMembers = append(chZMembers, redis.Z{Score: h.norm * 0.99, Member: h.id}) } chZSetKey := cache.HotRankZSetChannelKey(ch.ID) chTopIDsKey := cache.HotRankTopIDsChannelKey(ch.ID) if err := w.c.ZReplaceSortedSet(ctx, chZSetKey, chZMembers); err != nil { zap.L().Warn("hot rank channel ZReplaceSortedSet failed", zap.String("channel", ch.ID), zap.Error(err)) continue } w.c.Set(chTopIDsKey, chOrderedIDs, ttl) } return nil }