From 27ea8689f9e4c2a598389152877d9354f52eaff8 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Sun, 26 Apr 2026 11:39:41 +0800 Subject: [PATCH] feat(hot-rank): add per-channel hot ranking functionality 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. --- cmd/server/wire_gen.go | 2 +- internal/cache/keys.go | 8 ++ internal/handler/post_handler.go | 14 +-- internal/repository/post_repo.go | 102 +++++++++++++---- internal/service/hot_rank_worker.go | 171 +++++++++++++++++++++++++--- internal/service/post_service.go | 98 ++++++++++++---- internal/wire/service.go | 4 +- 7 files changed, 329 insertions(+), 70 deletions(-) diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index 2ae7d8e..18463af 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -144,7 +144,7 @@ func InitializeApp() (*App, error) { setupHandler := wire.ProvideSetupHandler(setupService) wsHandler := wire.ProvideWSHandler(hub, chatService, groupService, jwtService, callService, userRepository) router := ProvideRouter(userRepository, userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, callHandler, reportHandler, adminReportHandler, verificationHandler, adminVerificationHandler, adminProfileAuditHandler, setupHandler, logService, userActivityService, casbinService, wsHandler) - hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, cache) + hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, channelRepository, cache) app := NewApp(config, db, router, pushService, hotRankWorker, server, logger) return app, nil } diff --git a/internal/cache/keys.go b/internal/cache/keys.go index d9b8045..e9a5a2f 100644 --- a/internal/cache/keys.go +++ b/internal/cache/keys.go @@ -57,6 +57,14 @@ func HotRankTopIDsKey() string { return HotRankTopIDs } +func HotRankZSetChannelKey(channelID string) string { + return fmt.Sprintf("%s:ch:%s", HotRankZSet, channelID) +} + +func HotRankTopIDsChannelKey(channelID string) string { + return fmt.Sprintf("%s:ch:%s", HotRankTopIDs, channelID) +} + func PostListKey(postType string, userID string, page, pageSize int) string { if userID == "" { return fmt.Sprintf("%s:%s:%d:%d", PrefixPostList, postType, page, pageSize) diff --git a/internal/handler/post_handler.go b/internal/handler/post_handler.go index 5bac1a7..12a6128 100644 --- a/internal/handler/post_handler.go +++ b/internal/handler/post_handler.go @@ -293,24 +293,20 @@ func (h *PostHandler) List(c *gin.Context) { // 根据 tab 参数选择不同的获取方式 switch tab { case "follow": - // 获取关注用户的帖子，需要登录 if currentUserID == "" { response.Unauthorized(c, "请先登录") return } - posts, total, err = h.postService.GetFollowingPosts(c.Request.Context(), currentUserID, page, pageSize) + posts, total, err = h.postService.GetFollowingPosts(c.Request.Context(), currentUserID, page, pageSize, channelID) case "hot": - // 获取热门帖子 - posts, total, err = h.postService.GetHotPosts(c.Request.Context(), page, pageSize) + posts, total, err = h.postService.GetHotPosts(c.Request.Context(), page, pageSize, channelID) case "latest": - // 最新帖子 if userID != "" && userID == currentUserID { posts, total, err = h.postService.GetLatestPostsForOwner(c.Request.Context(), page, pageSize, userID, channelID) } else { posts, total, err = h.postService.GetLatestPosts(c.Request.Context(), page, pageSize, userID, channelID) } default: - // 默认获取最新帖子 if userID != "" && userID == currentUserID { posts, total, err = h.postService.GetLatestPostsForOwner(c.Request.Context(), page, pageSize, userID, channelID) } else { @@ -360,15 +356,13 @@ func (h *PostHandler) ListByCursor(c *gin.Context) { // 根据 tab 参数选择不同的获取方式 switch tab { case "follow": - // 获取关注用户的帖子，需要登录 if currentUserID == "" { response.Unauthorized(c, "请先登录") return } - result, err = h.postService.GetFollowingPostsByCursor(c.Request.Context(), currentUserID, req) + result, err = h.postService.GetFollowingPostsByCursor(c.Request.Context(), currentUserID, channelID, req) case "hot": - // 获取热门帖子 - result, err = h.postService.GetHotPostsByCursor(c.Request.Context(), req) + result, err = h.postService.GetHotPostsByCursor(c.Request.Context(), channelID, req) case "latest": // 最新帖子 result, err = h.postService.ListByCursor(c.Request.Context(), userID, includePending, channelID, req) diff --git a/internal/repository/post_repo.go b/internal/repository/post_repo.go index d70766b..39b08d0 100644 --- a/internal/repository/post_repo.go +++ b/internal/repository/post_repo.go @@ -33,10 +33,12 @@ type PostRepository interface { IncrementViews(postID string) error IncrementShares(postID string) (int, error) Search(keyword string, page, pageSize int) ([]*model.Post, int64, error) - GetFollowingPosts(userID string, page, pageSize int) ([]*model.Post, int64, error) - GetHotPosts(page, pageSize int) ([]*model.Post, int64, error) + GetFollowingPosts(userID string, page, pageSize int, channelID *string) ([]*model.Post, int64, error) + GetHotPosts(page, pageSize int, channelID *string) ([]*model.Post, int64, error) ListPublishedPostIDsSince(since time.Time, limit int) ([]string, error) + ListPublishedPostIDsByChannel(channelID string, recentHours int, limit int) ([]string, error) ListPinnedPublishedPostIDs() ([]string, error) + ListPinnedPublishedPostIDsByChannel(channelID string) ([]string, error) GetPostHotSnapshotsByIDs(ids []string) ([]PostHotSnapshot, error) GetByIDs(ids []string) ([]*model.Post, error) CreateWithContext(ctx context.Context, post *model.Post, images []string) error @@ -52,8 +54,8 @@ type PostRepository interface { GetPostsByCursor(ctx context.Context, userID string, includePending bool, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) SearchPostsByCursor(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) } // postRepository 帖子仓储实现 @@ -528,18 +530,21 @@ func (r *postRepository) Search(keyword string, page, pageSize int) ([]*model.Po } // GetFollowingPosts 获取关注用户的帖子 -func (r *postRepository) GetFollowingPosts(userID string, page, pageSize int) ([]*model.Post, int64, error) { +func (r *postRepository) GetFollowingPosts(userID string, page, pageSize int, channelID *string) ([]*model.Post, int64, error) { var posts []*model.Post var total int64 - // 子查询：获取当前用户关注的所有用户ID subQuery := r.db.Model(&model.Follow{}).Where("follower_id = ?", userID).Select("following_id") - // 统计总数 - r.db.Model(&model.Post{}).Where("user_id IN (?) AND status = ?", subQuery, model.PostStatusPublished).Count(&total) + baseQuery := r.db.Model(&model.Post{}).Where("user_id IN (?) AND status = ?", subQuery, model.PostStatusPublished) + if channelID != nil && *channelID != "" { + baseQuery = baseQuery.Where("channel_id = ?", *channelID) + } + + baseQuery.Count(&total) offset := (page - 1) * pageSize - err := r.db.Where("user_id IN (?) AND status = ?", subQuery, model.PostStatusPublished). + err := baseQuery. Preload("User").Preload("Images"). Offset(offset).Limit(pageSize). Order("created_at DESC"). @@ -549,14 +554,18 @@ func (r *postRepository) GetFollowingPosts(userID string, page, pageSize int) ([ } // GetHotPosts 热门榜降级：Redis 未就绪时按最新发布排序 -func (r *postRepository) GetHotPosts(page, pageSize int) ([]*model.Post, int64, error) { +func (r *postRepository) GetHotPosts(page, pageSize int, channelID *string) ([]*model.Post, int64, error) { var posts []*model.Post var total int64 - r.db.Model(&model.Post{}).Where("status = ?", model.PostStatusPublished).Count(&total) + baseQuery := r.db.Model(&model.Post{}).Where("status = ?", model.PostStatusPublished) + if channelID != nil && *channelID != "" { + baseQuery = baseQuery.Where("channel_id = ?", *channelID) + } + baseQuery.Count(&total) offset := (page - 1) * pageSize - err := r.db.Where("status = ?", model.PostStatusPublished).Preload("User").Preload("Images"). + err := baseQuery.Preload("User").Preload("Images"). Offset(offset).Limit(pageSize). Order("created_at DESC"). Find(&posts).Error @@ -625,6 +634,58 @@ func (r *postRepository) ListPinnedPublishedPostIDs() ([]string, error) { return out, nil } +func (r *postRepository) ListPublishedPostIDsByChannel(channelID string, recentHours int, limit int) ([]string, error) { + if limit <= 0 { + return nil, nil + } + type row struct { + ID string `gorm:"column:id"` + } + var rows []row + q := r.db.Model(&model.Post{}). + Select("id"). + Where("status = ? AND channel_id = ?", model.PostStatusPublished, channelID) + if recentHours > 0 { + since := time.Now().Add(-time.Duration(recentHours) * time.Hour) + q = q.Where("created_at >= ?", since) + } + err := q.Order("created_at DESC"). + Limit(limit). + Find(&rows).Error + if err != nil { + return nil, err + } + out := make([]string, 0, len(rows)) + for _, x := range rows { + if x.ID != "" { + out = append(out, x.ID) + } + } + return out, nil +} + +func (r *postRepository) ListPinnedPublishedPostIDsByChannel(channelID string) ([]string, error) { + type row struct { + ID string `gorm:"column:id"` + } + var rows []row + err := r.db.Model(&model.Post{}). + Select("id"). + Where("status = ? AND is_pinned = ? AND channel_id = ?", model.PostStatusPublished, true, channelID). + Order("created_at DESC"). + Find(&rows).Error + if err != nil { + return nil, err + } + out := make([]string, 0, len(rows)) + for _, x := range rows { + if x.ID != "" { + out = append(out, x.ID) + } + } + return out, nil +} + // GetPostHotSnapshotsByIDs 仅拉取候选 ID 对应的热度字段（分批 IN 查询） func (r *postRepository) GetPostHotSnapshotsByIDs(ids []string) ([]PostHotSnapshot, error) { if len(ids) == 0 { @@ -1187,14 +1248,15 @@ func (r *postRepository) GetUserPostsByCursor(ctx context.Context, userID string } // GetFollowingPostsByCursor 游标分页获取关注用户的帖子 -func (r *postRepository) GetFollowingPostsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) { +func (r *postRepository) GetFollowingPostsByCursor(ctx context.Context, userID string, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) { db := r.getDB(ctx).WithContext(ctx) - // 子查询：获取当前用户关注的所有用户ID subQuery := db.Model(&model.Follow{}).Where("follower_id = ?", userID).Select("following_id") - // 构建基础查询 query := db.Model(&model.Post{}).Where("user_id IN (?) AND status = ?", subQuery, model.PostStatusPublished) + if channelID != nil && *channelID != "" { + query = query.Where("channel_id = ?", *channelID) + } // 使用游标构建器 builder := cursor.NewBuilder(query, cursor.SortByCreatedAtDesc). @@ -1243,12 +1305,14 @@ func (r *postRepository) GetFollowingPostsByCursor(ctx context.Context, userID s } // GetHotPostsByCursor 游标分页获取热门帖子 -func (r *postRepository) GetHotPostsByCursor(ctx context.Context, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) { +func (r *postRepository) GetHotPostsByCursor(ctx context.Context, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) { db := r.getDB(ctx).WithContext(ctx) - // 热门游标降级：与 GetHotPosts 一致，按最新发布排序 - query := db.Model(&model.Post{}).Where("status = ?", model.PostStatusPublished). - Order("created_at DESC") + query := db.Model(&model.Post{}).Where("status = ?", model.PostStatusPublished) + if channelID != nil && *channelID != "" { + query = query.Where("channel_id = ?", *channelID) + } + query = query.Order("created_at DESC") // 使用游标构建器 builder := cursor.NewBuilder(query, cursor.SortByCreatedAtDesc). diff --git a/internal/service/hot_rank_worker.go b/internal/service/hot_rank_worker.go index e23e08b..df2a52d 100644 --- a/internal/service/hot_rank_worker.go +++ b/internal/service/hot_rank_worker.go @@ -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 +} diff --git a/internal/service/post_service.go b/internal/service/post_service.go index cdfb789..b37e5fc 100644 --- a/internal/service/post_service.go +++ b/internal/service/post_service.go @@ -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 } diff --git a/internal/wire/service.go b/internal/wire/service.go index a370b49..0c6846a 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -370,8 +370,8 @@ func ProvideLogService( } // ProvideQRCodeLoginService 提供二维码登录服务 -func ProvideHotRankWorker(cfg *config.Config, postRepo repository.PostRepository, cacheBackend cache.Cache) *service.HotRankWorker { - return service.NewHotRankWorker(cfg, postRepo, cacheBackend) +func ProvideHotRankWorker(cfg *config.Config, postRepo repository.PostRepository, channelRepo repository.ChannelRepository, cacheBackend cache.Cache) *service.HotRankWorker { + return service.NewHotRankWorker(cfg, postRepo, channelRepo, cacheBackend) } func ProvideQRCodeLoginService(