carrot_bbs/backend
4
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(hot-rank): add per-channel hot ranking functionality
Some checks failed
Build Backend / build (push) Failing after 1m52s
Details
Build Backend / build-docker (push) Has been skipped
Details

Browse Source 
Add channel-specific hot post filtering and ranking across the posts
listing flow. Introduce refreshChannelRanks to compute per-channel hot
scores, new Redis cache keys for channel hot ranks, and update
repository/service/handler layers to accept channelID filtering
parameters. Update HotRankWorker dependency injection to include
ChannelRepository.
This commit is contained in:
lafay
2026-04-26 11:39:41 +08:00
parent 02466603f9
commit 27ea8689f9
7 changed files with 329 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
internal/service/hot_rank_worker.go
View File

@@ -19,23 +19,24 @@ import (
// 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{}
cfg *config.Config
postRepo repository.PostRepository
channelRepo repository.ChannelRepository
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 {
func NewHotRankWorker(cfg *config.Config, postRepo repository.PostRepository, channelRepo repository.ChannelRepository, c cache.Cache) *HotRankWorker {
return &HotRankWorker{
cfg: cfg,
postRepo: postRepo,
c: c,
stopCh: make(chan struct{}),
doneCh: make(chan struct{}),
cfg: cfg,
postRepo: postRepo,
channelRepo: channelRepo,
c: c,
stopCh: make(chan struct{}),
doneCh: make(chan struct{}),
}
}
@@ -266,6 +267,11 @@ func (w *HotRankWorker) Refresh(ctx context.Context) error {
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
}
@@ -330,3 +336,140 @@ func hotRankBaseScore(likes, comments, favorites, shares, views int) float64 {
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 chSnapshots []repository.PostHotSnapshot
var chRanked []scored
filteredSnaps, ferr := w.postRepo.GetPostHotSnapshotsByIDs(chCandidateIDs)
if ferr != nil || len(filteredSnaps) == 0 {
continue
}
chSnapshots = filteredSnaps
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([]scored, len(chSnapshots))
for i := range chSnapshots {
var norm float64
if maxR > minR {
norm = (raw[i] - minR) / (maxR - minR)
} else {
norm = 1
}
chRanked[i] = scored{idx: i, norm: norm}
}
slices.SortFunc(chRanked, func(a, b scored) 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
}

98
internal/service/post_service.go
View File

@@ -44,12 +44,12 @@ type PostService interface {
ListByCursor(ctx context.Context, userID string, includePending bool, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
SearchByCursor(ctx context.Context, keyword string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
GetUserPostsByCursor(ctx context.Context, userID string, includePending bool, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
GetFollowingPostsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
GetHotPostsByCursor(ctx context.Context, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
GetFollowingPostsByCursor(ctx context.Context, userID string, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
GetHotPostsByCursor(ctx context.Context, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error)
// 关注和推荐
GetFollowingPosts(ctx context.Context, userID string, page, pageSize int) ([]*model.Post, int64, error)
GetHotPosts(ctx context.Context, page, pageSize int) ([]*model.Post, int64, error)
GetFollowingPosts(ctx context.Context, userID string, page, pageSize int, channelID *string) ([]*model.Post, int64, error)
GetHotPosts(ctx context.Context, page, pageSize int, channelID *string) ([]*model.Post, int64, error)
// 交互功能
Like(ctx context.Context, postID, userID string) error
@@ -560,7 +560,7 @@ func (s *postServiceImpl) Search(ctx context.Context, keyword string, page, page
}
// GetFollowingPosts 获取关注用户的帖子（带缓存）
func (s *postServiceImpl) GetFollowingPosts(ctx context.Context, userID string, page, pageSize int) ([]*model.Post, int64, error) {
func (s *postServiceImpl) GetFollowingPosts(ctx context.Context, userID string, page, pageSize int, channelID *string) ([]*model.Post, int64, error) {
cacheSettings := cache.GetSettings()
postListTTL := cacheSettings.PostListTTL
if postListTTL <= 0 {
@@ -575,8 +575,11 @@ func (s *postServiceImpl) GetFollowingPosts(ctx context.Context, userID string,
jitter = PostListJitterRatio
}
// 生成缓存键
cacheKey := cache.PostListKey("follow", userID, page, pageSize)
cacheSuffix := ""
if channelID != nil && *channelID != "" {
cacheSuffix = ":ch:" + *channelID
}
cacheKey := cache.PostListKey("follow", userID+cacheSuffix, page, pageSize)
result, err := cache.GetOrLoadTyped[*PostListResult](
s.cache,
@@ -585,7 +588,7 @@ func (s *postServiceImpl) GetFollowingPosts(ctx context.Context, userID string,
jitter,
nullTTL,
func() (*PostListResult, error) {
posts, total, err := s.postRepo.GetFollowingPosts(userID, page, pageSize)
posts, total, err := s.postRepo.GetFollowingPosts(userID, page, pageSize, channelID)
if err != nil {
return nil, err
}
@@ -602,8 +605,16 @@ func (s *postServiceImpl) GetFollowingPosts(ctx context.Context, userID string,
}
// GetHotPosts 获取热门帖子（优先分层缓存 top_ids → Redis ZSET，未就绪时回源 DB 按最新排序）
func (s *postServiceImpl) GetHotPosts(ctx context.Context, page, pageSize int) ([]*model.Post, int64, error) {
func (s *postServiceImpl) GetHotPosts(ctx context.Context, page, pageSize int, channelID *string) ([]*model.Post, int64, error) {
offset := (page - 1) * pageSize
if channelID != nil && *channelID != "" {
if posts, total, ok, err := s.tryHotPostsFromChannelCache(ctx, *channelID, offset, pageSize); ok {
return posts, total, err
} else if err != nil {
return nil, 0, err
}
return s.getHotPostsFromDB(ctx, page, pageSize, channelID)
}
if posts, total, ok, err := s.tryHotPostsFromTopIDsCache(offset, pageSize); ok {
return posts, total, err
} else if err != nil {
@@ -614,7 +625,42 @@ func (s *postServiceImpl) GetHotPosts(ctx context.Context, page, pageSize int) (
} else if err != nil {
return nil, 0, err
}
return s.getHotPostsFromDB(ctx, page, pageSize)
return s.getHotPostsFromDB(ctx, page, pageSize, nil)
}
func (s *postServiceImpl) tryHotPostsFromChannelCache(ctx context.Context, channelID string, offset, pageSize int) (posts []*model.Post, total int64, ok bool, err error) {
if s.cache == nil || pageSize <= 0 {
return nil, 0, false, nil
}
topIDsKey := cache.HotRankTopIDsChannelKey(channelID)
ids, hit := cache.GetTyped[[]string](s.cache, topIDsKey)
if hit && len(ids) > 0 {
if offset >= len(ids) {
return []*model.Post{}, int64(len(ids)), true, nil
}
end := offset + pageSize
if end > len(ids) {
end = len(ids)
}
posts, err = s.postRepo.GetByIDs(ids[offset:end])
if err != nil {
return nil, 0, false, err
}
return posts, int64(len(ids)), true, nil
}
zsetKey := cache.HotRankZSetChannelKey(channelID)
card, zerr := s.cache.ZCard(ctx, zsetKey)
if zerr == nil && card > 0 {
ids, zerr = s.cache.ZRevRange(ctx, zsetKey, int64(offset), int64(offset+pageSize-1))
if zerr == nil && len(ids) > 0 {
posts, err = s.postRepo.GetByIDs(ids)
if err != nil {
return nil, 0, false, err
}
return posts, card, true, nil
}
}
return nil, 0, false, nil
}
// tryHotPostsFromTopIDsCache 从热门榜有序 ID 列表取帖（命中 LayeredCache 本地层时可不访问 Redis）
@@ -664,8 +710,8 @@ func (s *postServiceImpl) tryHotPostsFromZSet(ctx context.Context, offset, pageS
}
// getHotPostsFromDB 热门未就绪时按最新发布降级
func (s *postServiceImpl) getHotPostsFromDB(ctx context.Context, page, pageSize int) ([]*model.Post, int64, error) {
posts, total, err := s.postRepo.GetHotPosts(page, pageSize)
func (s *postServiceImpl) getHotPostsFromDB(ctx context.Context, page, pageSize int, channelID *string) ([]*model.Post, int64, error) {
posts, total, err := s.postRepo.GetHotPosts(page, pageSize, channelID)
if err != nil {
return nil, 0, err
}
@@ -713,23 +759,21 @@ func (s *postServiceImpl) GetUserPostsByCursor(ctx context.Context, userID strin
}
// GetFollowingPostsByCursor 游标分页获取关注用户的帖子
func (s *postServiceImpl) GetFollowingPostsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
// 规范化请求参数
func (s *postServiceImpl) GetFollowingPostsByCursor(ctx context.Context, userID string, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
req.Normalize()
return s.postRepo.GetFollowingPostsByCursor(ctx, userID, req)
return s.postRepo.GetFollowingPostsByCursor(ctx, userID, channelID, req)
}
// GetHotPostsByCursor 游标分页获取热门帖子
func (s *postServiceImpl) GetHotPostsByCursor(ctx context.Context, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
// 规范化请求参数
func (s *postServiceImpl) GetHotPostsByCursor(ctx context.Context, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
req.Normalize()
return s.getHotPostsByCursorFromDB(ctx, req)
return s.getHotPostsByCursorFromDB(ctx, channelID, req)
}
// getHotPostsByCursorFromDB 热门游标：优先 ZSET（一次多取 1 条判断 hasMore），否则回源 DB
func (s *postServiceImpl) getHotPostsByCursorFromDB(ctx context.Context, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
func (s *postServiceImpl) getHotPostsByCursorFromDB(ctx context.Context, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
offset := 0
if req.Cursor != "" {
if parsed, err := strconv.Atoi(req.Cursor); err == nil && parsed >= 0 {
@@ -738,7 +782,14 @@ func (s *postServiceImpl) getHotPostsByCursorFromDB(ctx context.Context, req *cu
}
if s.cache != nil {
if ids, hit := cache.GetTyped[[]string](s.cache, cache.HotRankTopIDsKey()); hit && len(ids) > 0 {
topIDsKey := cache.HotRankTopIDsKey()
zsetKey := cache.HotRankZSetKey()
if channelID != nil && *channelID != "" {
topIDsKey = cache.HotRankTopIDsChannelKey(*channelID)
zsetKey = cache.HotRankZSetChannelKey(*channelID)
}
if ids, hit := cache.GetTyped[[]string](s.cache, topIDsKey); hit && len(ids) > 0 {
if offset < len(ids) {
stop := offset + req.PageSize + 1
if stop > len(ids) {
@@ -772,11 +823,10 @@ func (s *postServiceImpl) getHotPostsByCursorFromDB(ctx context.Context, req *cu
}, nil
}
key := cache.HotRankZSetKey()
card, zerr := s.cache.ZCard(ctx, key)
card, zerr := s.cache.ZCard(ctx, zsetKey)
if zerr == nil && card > 0 {
stop := int64(offset + req.PageSize)
ids, rerr := s.cache.ZRevRange(ctx, key, int64(offset), stop)
ids, rerr := s.cache.ZRevRange(ctx, zsetKey, int64(offset), stop)
if rerr == nil && len(ids) > 0 {
hasMore := len(ids) > req.PageSize
if hasMore {
@@ -801,7 +851,7 @@ func (s *postServiceImpl) getHotPostsByCursorFromDB(ctx context.Context, req *cu
}
page := offset/req.PageSize + 1
posts, _, err := s.postRepo.GetHotPosts(page, req.PageSize+1)
posts, _, err := s.postRepo.GetHotPosts(page, req.PageSize+1, channelID)
if err != nil {
return nil, err
}