refactor: remove Gorse integration and implement HotRank feature
- Removed Gorse-related configurations, handlers, and dependencies from the codebase. - Introduced HotRank feature with configuration options for ranking posts based on recent activity. - Updated application structure to support HotRank processing, including new caching mechanisms and database interactions. - Cleaned up related DTOs and repository methods to reflect the removal of Gorse and the addition of HotRank functionality.
This commit is contained in:
lafay
335
internal/service/hot_rank_worker.go
Normal file
335
internal/service/hot_rank_worker.go
Normal file
@@ -0,0 +1,335 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"math"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"carrot_bbs/internal/cache"
|
||||
"carrot_bbs/internal/config"
|
||||
"carrot_bbs/internal/repository"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// HotRankWorker 定时在候选池上重算热门分，写入 Redis ZSET（仅 TopN）及 top_ids（Redis+本地缓存）
|
||||
type HotRankWorker struct {
|
||||
cfg *config.Config
|
||||
postRepo *repository.PostRepository
|
||||
c cache.Cache
|
||||
mu sync.Mutex
|
||||
stopOnce sync.Once
|
||||
stopCh chan struct{}
|
||||
doneCh chan struct{}
|
||||
}
|
||||
|
||||
// NewHotRankWorker 创建热门榜重算 worker（cache 需为 Redis 实现才有意义）
|
||||
func NewHotRankWorker(cfg *config.Config, postRepo *repository.PostRepository, c cache.Cache) *HotRankWorker {
|
||||
return &HotRankWorker{
|
||||
cfg: cfg,
|
||||
postRepo: postRepo,
|
||||
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
|
||||
}
|
||||
if topN > 500 {
|
||||
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
|
||||
}
|
||||
|
||||
type scored struct {
|
||||
idx int
|
||||
norm float64
|
||||
}
|
||||
|
||||
var snapshots []repository.PostHotSnapshot
|
||||
var ranked []scored
|
||||
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, maxR := raw[0], raw[0]
|
||||
for _, v := range raw[1:] {
|
||||
if v < minR {
|
||||
minR = v
|
||||
}
|
||||
if v > maxR {
|
||||
maxR = v
|
||||
}
|
||||
}
|
||||
|
||||
ranked = make([]scored, len(snapshots))
|
||||
for i := range snapshots {
|
||||
var norm float64
|
||||
if maxR > minR {
|
||||
norm = (raw[i] - minR) / (maxR - minR)
|
||||
} else {
|
||||
norm = 1
|
||||
}
|
||||
ranked[i] = scored{idx: i, norm: norm}
|
||||
}
|
||||
sort.Slice(ranked, func(a, b int) bool {
|
||||
if ranked[a].norm != ranked[b].norm {
|
||||
return ranked[a].norm > ranked[b].norm
|
||||
}
|
||||
return snapshots[ranked[a].idx].CreatedAt.After(snapshots[ranked[b].idx].CreatedAt)
|
||||
})
|
||||
}
|
||||
|
||||
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)),
|
||||
)
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user