05c7f8a5eb
Add rich message segment support enabling structured content (text, images, votes, mentions) in posts and comments. Includes segments field in models, DTOs, handlers, and services. Also adds @mention notification processing and migrates vote data to embedded segments.
826 lines
26 KiB
Go
826 lines
26 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"with_you/internal/cache"
|
||
"with_you/internal/dto"
|
||
"with_you/internal/model"
|
||
"with_you/internal/pkg/cursor"
|
||
"with_you/internal/pkg/hook"
|
||
"with_you/internal/repository"
|
||
)
|
||
|
||
// 缓存TTL常量
|
||
const (
|
||
PostListTTL = 30 * time.Second // 帖子列表缓存30秒
|
||
PostListNullTTL = 5 * time.Second
|
||
PostListJitterRatio = 0.15
|
||
)
|
||
|
||
// PostService 帖子服务接口
|
||
type PostService interface {
|
||
// 帖子CRUD
|
||
Create(ctx context.Context, userID, title, content string, segments model.MessageSegments, images []string, channelID *string) (*model.Post, error)
|
||
GetByID(ctx context.Context, id string) (*model.Post, error)
|
||
Update(ctx context.Context, post *model.Post) error
|
||
UpdateWithImages(ctx context.Context, post *model.Post, images *[]string) error
|
||
Delete(ctx context.Context, id string) error
|
||
|
||
// 帖子列表
|
||
List(ctx context.Context, page, pageSize int, userID string, includePending bool, channelID *string) ([]*model.Post, int64, error)
|
||
GetLatestPosts(ctx context.Context, page, pageSize int, userID string, channelID *string) ([]*model.Post, int64, error)
|
||
GetLatestPostsForOwner(ctx context.Context, page, pageSize int, userID string, channelID *string) ([]*model.Post, int64, error)
|
||
GetUserPosts(ctx context.Context, userID string, page, pageSize int, includePending bool) ([]*model.Post, int64, error)
|
||
GetFavorites(ctx context.Context, userID string, page, pageSize int) ([]*model.Post, int64, error)
|
||
Search(ctx context.Context, keyword string, page, pageSize int) ([]*model.Post, int64, error)
|
||
|
||
// 游标分页方法
|
||
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)
|
||
|
||
// 关注和推荐
|
||
GetFollowingPosts(ctx context.Context, userID string, page, pageSize int) ([]*model.Post, int64, error)
|
||
GetHotPosts(ctx context.Context, page, pageSize int) ([]*model.Post, int64, error)
|
||
|
||
// 交互功能
|
||
Like(ctx context.Context, postID, userID string) error
|
||
Unlike(ctx context.Context, postID, userID string) error
|
||
IsLiked(ctx context.Context, postID, userID string) bool
|
||
Favorite(ctx context.Context, postID, userID string) error
|
||
Unfavorite(ctx context.Context, postID, userID string) error
|
||
IsFavorited(ctx context.Context, postID, userID string) bool
|
||
|
||
// 交互状态查询
|
||
GetPostInteractionStatus(ctx context.Context, postIDs []string, userID string) (map[string]bool, map[string]bool, error)
|
||
GetPostInteractionStatusSingle(ctx context.Context, postID, userID string) (isLiked, isFavorited bool)
|
||
|
||
// 其他
|
||
IncrementViews(ctx context.Context, postID, userID string) error
|
||
// RecordShare 记录分享（仅已发布帖子计数 +1），返回最新 shares_count
|
||
RecordShare(ctx context.Context, postID, userID string) (sharesCount int, err error)
|
||
}
|
||
|
||
// postServiceImpl 帖子服务实现
|
||
type postServiceImpl struct {
|
||
postRepo repository.PostRepository
|
||
systemMessageService SystemMessageService
|
||
cache cache.Cache
|
||
postAIService *PostAIService
|
||
txManager repository.TransactionManager
|
||
logService *LogService
|
||
hookManager *hook.Manager
|
||
}
|
||
|
||
func NewPostService(postRepo repository.PostRepository, systemMessageService SystemMessageService, postAIService *PostAIService, cacheBackend cache.Cache, txManager repository.TransactionManager, hookManager *hook.Manager, logService *LogService) PostService {
|
||
return &postServiceImpl{
|
||
postRepo: postRepo,
|
||
systemMessageService: systemMessageService,
|
||
cache: cacheBackend,
|
||
postAIService: postAIService,
|
||
txManager: txManager,
|
||
logService: logService,
|
||
hookManager: hookManager,
|
||
}
|
||
}
|
||
|
||
// PostListResult 帖子列表缓存结果
|
||
type PostListResult struct {
|
||
Posts []*model.Post
|
||
Total int64
|
||
}
|
||
|
||
// Create 创建帖子
|
||
func (s *postServiceImpl) Create(ctx context.Context, userID, title, content string, segments model.MessageSegments, images []string, channelID *string) (*model.Post, error) {
|
||
// 如果没有提供 content 但有 segments，从 segments 提取文本摘要
|
||
if content == "" && len(segments) > 0 {
|
||
content = dto.ExtractTextContent(segments)
|
||
}
|
||
// 如果有 segments 但 content 为空，将 content 设为空字符串（NOT NULL 约束）
|
||
if content == "" {
|
||
content = " "
|
||
}
|
||
|
||
// 检查 segments 中是否包含投票
|
||
isVote := dto.HasVoteSegment(segments)
|
||
|
||
post := &model.Post{
|
||
UserID: userID,
|
||
ChannelID: channelID,
|
||
Title: title,
|
||
Content: content,
|
||
Segments: segments,
|
||
Status: model.PostStatusPending,
|
||
IsVote: isVote,
|
||
}
|
||
|
||
err := s.postRepo.Create(post, images)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 记录操作日志
|
||
if s.logService != nil {
|
||
s.logService.OperationLog.RecordOperation(&model.OperationLog{
|
||
UserID: userID,
|
||
Operation: string(model.OpPostCreate),
|
||
Module: "post",
|
||
TargetType: "post",
|
||
TargetID: post.ID,
|
||
Status: "success",
|
||
})
|
||
}
|
||
|
||
// 失效帖子列表缓存
|
||
cache.InvalidatePostList(s.cache)
|
||
|
||
// 异步处理 @提及通知
|
||
if len(segments) > 0 {
|
||
bgCtx := context.WithoutCancel(ctx)
|
||
go s.processMentionNotifications(bgCtx, userID, post.ID, post.Title, segments)
|
||
}
|
||
|
||
// 异步执行审核流
|
||
go s.reviewPostAsync(post.ID, userID, title, content, images)
|
||
|
||
// 重新查询以获取关联的 User 和 Images
|
||
return s.postRepo.GetByID(post.ID)
|
||
}
|
||
|
||
func (s *postServiceImpl) reviewPostAsync(postID, userID, title, content string, images []string) {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[ERROR] Panic in post moderation async flow, fallback publish post=%s panic=%v", postID, r)
|
||
if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "system"); err != nil {
|
||
log.Printf("[WARN] Failed to publish post %s after panic recovery: %v", postID, err)
|
||
return
|
||
}
|
||
s.invalidatePostCaches(postID)
|
||
}
|
||
}()
|
||
|
||
if s.hookManager == nil {
|
||
if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "system"); err != nil {
|
||
log.Printf("[WARN] Failed to publish post without hook manager: %v", err)
|
||
} else {
|
||
s.invalidatePostCaches(postID)
|
||
}
|
||
return
|
||
}
|
||
|
||
var authorID uint
|
||
fmt.Sscanf(userID, "%d", &authorID)
|
||
|
||
result := &hook.ModerationResult{}
|
||
s.hookManager.TriggerWithMetadata(context.Background(), hook.HookPostPreModerate, authorID, &hook.PostModerateHookData{
|
||
PostID: postID,
|
||
Title: title,
|
||
Content: content,
|
||
Images: images,
|
||
AuthorID: authorID,
|
||
}, map[string]any{
|
||
"result": result,
|
||
})
|
||
|
||
s.hookManager.Trigger(context.Background(), hook.HookPostModerated, authorID, &hook.PostModeratedHookData{
|
||
PostID: postID,
|
||
AuthorID: authorID,
|
||
Approved: result.Approved,
|
||
RejectReason: result.RejectReason,
|
||
ReviewedBy: result.ReviewedBy,
|
||
})
|
||
|
||
if !result.Approved {
|
||
if result.NeedsReview {
|
||
s.invalidatePostCaches(postID)
|
||
return
|
||
}
|
||
if updateErr := s.updateModerationStatusWithRetry(postID, model.PostStatusRejected, result.RejectReason, result.ReviewedBy); updateErr != nil {
|
||
log.Printf("[WARN] Failed to reject post %s: %v", postID, updateErr)
|
||
} else {
|
||
s.invalidatePostCaches(postID)
|
||
}
|
||
s.notifyModerationRejected(userID, result.RejectReason)
|
||
return
|
||
}
|
||
|
||
if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", result.ReviewedBy); err != nil {
|
||
log.Printf("[WARN] Failed to publish post %s: %v", postID, err)
|
||
return
|
||
}
|
||
s.invalidatePostCaches(postID)
|
||
|
||
}
|
||
|
||
func (s *postServiceImpl) updateModerationStatusWithRetry(postID string, status model.PostStatus, rejectReason string, reviewedBy string) error {
|
||
const maxAttempts = 3
|
||
const retryDelay = 200 * time.Millisecond
|
||
|
||
var lastErr error
|
||
for attempt := 1; attempt <= maxAttempts; attempt++ {
|
||
if err := s.postRepo.UpdateModerationStatus(postID, status, rejectReason, reviewedBy); err != nil {
|
||
lastErr = err
|
||
if attempt < maxAttempts {
|
||
log.Printf("[WARN] UpdateModerationStatus failed post=%s attempt=%d/%d err=%v", postID, attempt, maxAttempts, err)
|
||
time.Sleep(time.Duration(attempt) * retryDelay)
|
||
continue
|
||
}
|
||
} else {
|
||
return nil
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
func (s *postServiceImpl) invalidatePostCaches(postID string) {
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
}
|
||
|
||
func (s *postServiceImpl) notifyModerationRejected(userID, reason string) {
|
||
if s.systemMessageService == nil || strings.TrimSpace(userID) == "" {
|
||
return
|
||
}
|
||
|
||
content := "您发布的帖子未通过AI审核，请修改后重试。"
|
||
if strings.TrimSpace(reason) != "" {
|
||
content = fmt.Sprintf("您发布的帖子未通过AI审核，原因：%s。请修改后重试。", reason)
|
||
}
|
||
|
||
go func() {
|
||
if err := s.systemMessageService.SendSystemAnnouncement(
|
||
context.Background(),
|
||
[]string{userID},
|
||
"帖子审核未通过",
|
||
content,
|
||
); err != nil {
|
||
log.Printf("[WARN] Failed to send moderation reject notification: %v", err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
// processMentionNotifications 处理帖子中 @提及的通知
|
||
func (s *postServiceImpl) processMentionNotifications(ctx context.Context, authorID, postID, postTitle string, segments model.MessageSegments) {
|
||
if s.systemMessageService == nil {
|
||
return
|
||
}
|
||
|
||
mentionedUserIDs := dto.ExtractMentionedUsers(segments)
|
||
if len(mentionedUserIDs) == 0 {
|
||
return
|
||
}
|
||
|
||
for _, targetUserID := range mentionedUserIDs {
|
||
if targetUserID == authorID {
|
||
continue
|
||
}
|
||
if err := s.systemMessageService.SendMentionNotification(ctx, targetUserID, authorID, postID); err != nil {
|
||
log.Printf("[WARN] Failed to send mention notification to user %s: %v", targetUserID, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// GetByID 根据ID获取帖子
|
||
func (s *postServiceImpl) GetByID(ctx context.Context, id string) (*model.Post, error) {
|
||
return s.postRepo.GetByID(id)
|
||
}
|
||
|
||
// Update 更新帖子
|
||
func (s *postServiceImpl) Update(ctx context.Context, post *model.Post) error {
|
||
return s.UpdateWithImages(ctx, post, nil)
|
||
}
|
||
|
||
// UpdateWithImages 更新帖子并可选更新图片（images=nil 表示不更新图片）
|
||
func (s *postServiceImpl) UpdateWithImages(ctx context.Context, post *model.Post, images *[]string) error {
|
||
err := s.postRepo.UpdateWithImages(post, images)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存和列表缓存
|
||
cache.InvalidatePostDetail(s.cache, post.ID)
|
||
cache.InvalidatePostList(s.cache)
|
||
|
||
return nil
|
||
}
|
||
|
||
// Delete 删除帖子
|
||
func (s *postServiceImpl) Delete(ctx context.Context, id string) error {
|
||
err := s.postRepo.Delete(id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存和列表缓存
|
||
cache.InvalidatePostDetail(s.cache, id)
|
||
cache.InvalidatePostList(s.cache)
|
||
|
||
return nil
|
||
}
|
||
|
||
// List 获取帖子列表（带缓存）
|
||
func (s *postServiceImpl) List(ctx context.Context, page, pageSize int, userID string, includePending bool, channelID *string) ([]*model.Post, int64, error) {
|
||
cacheSettings := cache.GetSettings()
|
||
postListTTL := cacheSettings.PostListTTL
|
||
if postListTTL <= 0 {
|
||
postListTTL = PostListTTL
|
||
}
|
||
nullTTL := cacheSettings.NullTTL
|
||
if nullTTL <= 0 {
|
||
nullTTL = PostListNullTTL
|
||
}
|
||
jitter := cacheSettings.JitterRatio
|
||
if jitter <= 0 {
|
||
jitter = PostListJitterRatio
|
||
}
|
||
|
||
// 生成缓存键（包含 userID 维度与可见性维度，避免作者视角污染公开视角）
|
||
visibilityUserKey := userID
|
||
if includePending && userID != "" {
|
||
visibilityUserKey = "owner:" + userID
|
||
}
|
||
if channelID != nil && *channelID != "" {
|
||
visibilityUserKey += ":channel:" + *channelID
|
||
}
|
||
cacheKey := cache.PostListKey("latest", visibilityUserKey, page, pageSize)
|
||
|
||
result, err := cache.GetOrLoadTyped(
|
||
s.cache,
|
||
cacheKey,
|
||
postListTTL,
|
||
jitter,
|
||
nullTTL,
|
||
func() (*PostListResult, error) {
|
||
posts, total, err := s.postRepo.List(page, pageSize, userID, includePending, channelID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &PostListResult{Posts: posts, Total: total}, nil
|
||
},
|
||
)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if result == nil {
|
||
return []*model.Post{}, 0, nil
|
||
}
|
||
|
||
// 兼容历史脏缓存：旧缓存序列化会丢失 Post.User，导致前端显示“匿名用户”
|
||
// 这里检测并回源重建一次缓存，避免在 TTL 内持续返回缺失作者的数据。
|
||
missingAuthor := false
|
||
for _, post := range result.Posts {
|
||
if post != nil && post.UserID != "" && post.User == nil {
|
||
missingAuthor = true
|
||
break
|
||
}
|
||
}
|
||
if missingAuthor {
|
||
posts, total, loadErr := s.postRepo.List(page, pageSize, userID, includePending, channelID)
|
||
if loadErr != nil {
|
||
return nil, 0, loadErr
|
||
}
|
||
result = &PostListResult{Posts: posts, Total: total}
|
||
cache.SetWithJitter(s.cache, cacheKey, result, postListTTL, jitter)
|
||
}
|
||
|
||
return result.Posts, result.Total, nil
|
||
}
|
||
|
||
// GetLatestPosts 获取最新帖子（语义化别名）
|
||
func (s *postServiceImpl) GetLatestPosts(ctx context.Context, page, pageSize int, userID string, channelID *string) ([]*model.Post, int64, error) {
|
||
return s.List(ctx, page, pageSize, userID, false, channelID)
|
||
}
|
||
|
||
// GetLatestPostsForOwner 获取作者视角帖子列表（包含待审核）
|
||
func (s *postServiceImpl) GetLatestPostsForOwner(ctx context.Context, page, pageSize int, userID string, channelID *string) ([]*model.Post, int64, error) {
|
||
return s.List(ctx, page, pageSize, userID, true, channelID)
|
||
}
|
||
|
||
// GetUserPosts 获取用户帖子
|
||
func (s *postServiceImpl) GetUserPosts(ctx context.Context, userID string, page, pageSize int, includePending bool) ([]*model.Post, int64, error) {
|
||
return s.postRepo.GetUserPosts(userID, page, pageSize, includePending)
|
||
}
|
||
|
||
// Like 点赞
|
||
func (s *postServiceImpl) Like(ctx context.Context, postID, userID string) error {
|
||
// 获取帖子信息用于发送通知
|
||
post, err := s.postRepo.GetByID(postID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
err = s.postRepo.Like(postID, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
|
||
// 发送点赞通知（不给自己发通知）
|
||
if s.systemMessageService != nil && post.UserID != userID {
|
||
go func() {
|
||
notifyErr := s.systemMessageService.SendLikeNotification(context.Background(), post.UserID, userID, postID)
|
||
if notifyErr != nil {
|
||
log.Printf("[ERROR] Error sending like notification: %v", notifyErr)
|
||
}
|
||
}()
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// Unlike 取消点赞
|
||
func (s *postServiceImpl) Unlike(ctx context.Context, postID, userID string) error {
|
||
err := s.postRepo.Unlike(postID, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
|
||
return nil
|
||
}
|
||
|
||
// IsLiked 检查是否点赞
|
||
func (s *postServiceImpl) IsLiked(ctx context.Context, postID, userID string) bool {
|
||
return s.postRepo.IsLiked(postID, userID)
|
||
}
|
||
|
||
// Favorite 收藏
|
||
func (s *postServiceImpl) Favorite(ctx context.Context, postID, userID string) error {
|
||
// 获取帖子信息用于发送通知
|
||
post, err := s.postRepo.GetByID(postID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
err = s.postRepo.Favorite(postID, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
|
||
// 发送收藏通知（不给自己发通知）
|
||
if s.systemMessageService != nil && post.UserID != userID {
|
||
go func() {
|
||
notifyErr := s.systemMessageService.SendFavoriteNotification(context.Background(), post.UserID, userID, postID)
|
||
if notifyErr != nil {
|
||
log.Printf("[ERROR] Error sending favorite notification: %v", notifyErr)
|
||
}
|
||
}()
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// Unfavorite 取消收藏
|
||
func (s *postServiceImpl) Unfavorite(ctx context.Context, postID, userID string) error {
|
||
err := s.postRepo.Unfavorite(postID, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
|
||
return nil
|
||
}
|
||
|
||
// IsFavorited 检查是否收藏
|
||
func (s *postServiceImpl) IsFavorited(ctx context.Context, postID, userID string) bool {
|
||
return s.postRepo.IsFavorited(postID, userID)
|
||
}
|
||
|
||
// GetPostInteractionStatus 批量获取帖子的交互状态（点赞、收藏）
|
||
func (s *postServiceImpl) GetPostInteractionStatus(ctx context.Context, postIDs []string, userID string) (map[string]bool, map[string]bool, error) {
|
||
isLikedMap := make(map[string]bool)
|
||
isFavoritedMap := make(map[string]bool)
|
||
|
||
if userID == "" || len(postIDs) == 0 {
|
||
return isLikedMap, isFavoritedMap, nil
|
||
}
|
||
|
||
// 使用批量查询替代循环中的单个查询，解决 N+1 问题
|
||
isLikedMap = s.postRepo.IsLikedBatch(postIDs, userID)
|
||
isFavoritedMap = s.postRepo.IsFavoritedBatch(postIDs, userID)
|
||
|
||
return isLikedMap, isFavoritedMap, nil
|
||
}
|
||
|
||
// GetPostInteractionStatusSingle 获取单个帖子的交互状态
|
||
func (s *postServiceImpl) GetPostInteractionStatusSingle(ctx context.Context, postID, userID string) (isLiked, isFavorited bool) {
|
||
if userID == "" {
|
||
return false, false
|
||
}
|
||
return s.postRepo.IsLiked(postID, userID), s.postRepo.IsFavorited(postID, userID)
|
||
}
|
||
|
||
// IncrementViews 增加帖子观看量
|
||
func (s *postServiceImpl) IncrementViews(ctx context.Context, postID, userID string) error {
|
||
if err := s.postRepo.IncrementViews(postID); err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// RecordShare 记录分享：仅已发布帖子增加 shares_count，并失效帖子详情与列表缓存
|
||
func (s *postServiceImpl) RecordShare(ctx context.Context, postID, userID string) (int, error) {
|
||
n, err := s.postRepo.IncrementShares(postID)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if s.cache != nil {
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// GetFavorites 获取收藏列表
|
||
func (s *postServiceImpl) GetFavorites(ctx context.Context, userID string, page, pageSize int) ([]*model.Post, int64, error) {
|
||
return s.postRepo.GetFavorites(userID, page, pageSize)
|
||
}
|
||
|
||
// Search 搜索帖子
|
||
func (s *postServiceImpl) Search(ctx context.Context, keyword string, page, pageSize int) ([]*model.Post, int64, error) {
|
||
return s.postRepo.Search(keyword, page, pageSize)
|
||
}
|
||
|
||
// GetFollowingPosts 获取关注用户的帖子（带缓存）
|
||
func (s *postServiceImpl) GetFollowingPosts(ctx context.Context, userID string, page, pageSize int) ([]*model.Post, int64, error) {
|
||
cacheSettings := cache.GetSettings()
|
||
postListTTL := cacheSettings.PostListTTL
|
||
if postListTTL <= 0 {
|
||
postListTTL = PostListTTL
|
||
}
|
||
nullTTL := cacheSettings.NullTTL
|
||
if nullTTL <= 0 {
|
||
nullTTL = PostListNullTTL
|
||
}
|
||
jitter := cacheSettings.JitterRatio
|
||
if jitter <= 0 {
|
||
jitter = PostListJitterRatio
|
||
}
|
||
|
||
// 生成缓存键
|
||
cacheKey := cache.PostListKey("follow", userID, page, pageSize)
|
||
|
||
result, err := cache.GetOrLoadTyped[*PostListResult](
|
||
s.cache,
|
||
cacheKey,
|
||
postListTTL,
|
||
jitter,
|
||
nullTTL,
|
||
func() (*PostListResult, error) {
|
||
posts, total, err := s.postRepo.GetFollowingPosts(userID, page, pageSize)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &PostListResult{Posts: posts, Total: total}, nil
|
||
},
|
||
)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if result == nil {
|
||
return []*model.Post{}, 0, nil
|
||
}
|
||
return result.Posts, result.Total, nil
|
||
}
|
||
|
||
// GetHotPosts 获取热门帖子（优先分层缓存 top_ids → Redis ZSET，未就绪时回源 DB 按最新排序）
|
||
func (s *postServiceImpl) GetHotPosts(ctx context.Context, page, pageSize int) ([]*model.Post, int64, error) {
|
||
offset := (page - 1) * pageSize
|
||
if posts, total, ok, err := s.tryHotPostsFromTopIDsCache(offset, pageSize); ok {
|
||
return posts, total, err
|
||
} else if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if posts, total, ok, err := s.tryHotPostsFromZSet(ctx, offset, pageSize); ok {
|
||
return posts, total, err
|
||
} else if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return s.getHotPostsFromDB(ctx, page, pageSize)
|
||
}
|
||
|
||
// tryHotPostsFromTopIDsCache 从热门榜有序 ID 列表取帖（命中 LayeredCache 本地层时可不访问 Redis）
|
||
func (s *postServiceImpl) tryHotPostsFromTopIDsCache(offset, pageSize int) (posts []*model.Post, total int64, ok bool, err error) {
|
||
if s.cache == nil || pageSize <= 0 {
|
||
return nil, 0, false, nil
|
||
}
|
||
ids, hit := cache.GetTyped[[]string](s.cache, cache.HotRankTopIDsKey())
|
||
if !hit || len(ids) == 0 {
|
||
return nil, 0, false, nil
|
||
}
|
||
if offset >= len(ids) {
|
||
return []*model.Post{}, int64(len(ids)), true, nil
|
||
}
|
||
end := offset + pageSize
|
||
if end > len(ids) {
|
||
end = len(ids)
|
||
}
|
||
pageIDs := ids[offset:end]
|
||
posts, err = s.postRepo.GetByIDs(pageIDs)
|
||
if err != nil {
|
||
return nil, 0, false, err
|
||
}
|
||
return posts, int64(len(ids)), true, nil
|
||
}
|
||
|
||
// tryHotPostsFromZSet 从定时任务写入的 ZSET 按排名取帖；ok=false 表示应回源
|
||
func (s *postServiceImpl) tryHotPostsFromZSet(ctx context.Context, offset, pageSize int) (posts []*model.Post, total int64, ok bool, err error) {
|
||
if s.cache == nil || pageSize <= 0 {
|
||
return nil, 0, false, nil
|
||
}
|
||
key := cache.HotRankZSetKey()
|
||
card, zerr := s.cache.ZCard(ctx, key)
|
||
if zerr != nil || card == 0 {
|
||
return nil, 0, false, nil
|
||
}
|
||
stop := int64(offset + pageSize - 1)
|
||
ids, zerr := s.cache.ZRevRange(ctx, key, int64(offset), stop)
|
||
if zerr != nil || len(ids) == 0 {
|
||
return nil, 0, false, nil
|
||
}
|
||
posts, err = s.postRepo.GetByIDs(ids)
|
||
if err != nil {
|
||
return nil, 0, false, err
|
||
}
|
||
return posts, card, true, nil
|
||
}
|
||
|
||
// getHotPostsFromDB 热门未就绪时按最新发布降级
|
||
func (s *postServiceImpl) getHotPostsFromDB(ctx context.Context, page, pageSize int) ([]*model.Post, int64, error) {
|
||
posts, total, err := s.postRepo.GetHotPosts(page, pageSize)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return posts, total, nil
|
||
}
|
||
|
||
// ========== 事务管理器示例方法 ==========
|
||
|
||
// DeletePostWithTransaction 使用事务管理器删除帖子（示例）
|
||
// 此方法展示如何在 Service 层使用事务管理器控制跨多个 Repository 的事务
|
||
// 当需要在一个事务中执行多个 Repository 操作时，可以使用此模式
|
||
func (s *postServiceImpl) DeletePostWithTransaction(ctx context.Context, postID string) error {
|
||
// 使用事务管理器执行事务
|
||
return s.txManager.RunInTransaction(ctx, func(ctx context.Context) error {
|
||
// 在同一个事务中执行删除操作
|
||
// Repository 方法会通过 context 获取事务 TX
|
||
return s.postRepo.DeleteWithContext(ctx, postID)
|
||
})
|
||
}
|
||
|
||
// ========== 游标分页方法 ==========
|
||
|
||
// ListByCursor 游标分页获取帖子列表
|
||
func (s *postServiceImpl) ListByCursor(ctx context.Context, userID string, includePending bool, channelID *string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
|
||
// 规范化请求参数
|
||
req.Normalize()
|
||
|
||
return s.postRepo.GetPostsByCursor(ctx, userID, includePending, channelID, req)
|
||
}
|
||
|
||
// SearchByCursor 游标分页搜索帖子
|
||
func (s *postServiceImpl) SearchByCursor(ctx context.Context, keyword string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
|
||
// 规范化请求参数
|
||
req.Normalize()
|
||
|
||
return s.postRepo.SearchPostsByCursor(ctx, keyword, req)
|
||
}
|
||
|
||
// GetUserPostsByCursor 游标分页获取用户帖子
|
||
func (s *postServiceImpl) GetUserPostsByCursor(ctx context.Context, userID string, includePending bool, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
|
||
// 规范化请求参数
|
||
req.Normalize()
|
||
|
||
return s.postRepo.GetUserPostsByCursor(ctx, userID, includePending, req)
|
||
}
|
||
|
||
// GetFollowingPostsByCursor 游标分页获取关注用户的帖子
|
||
func (s *postServiceImpl) GetFollowingPostsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
|
||
// 规范化请求参数
|
||
req.Normalize()
|
||
|
||
return s.postRepo.GetFollowingPostsByCursor(ctx, userID, req)
|
||
}
|
||
|
||
// GetHotPostsByCursor 游标分页获取热门帖子
|
||
func (s *postServiceImpl) GetHotPostsByCursor(ctx context.Context, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
|
||
// 规范化请求参数
|
||
req.Normalize()
|
||
|
||
return s.getHotPostsByCursorFromDB(ctx, req)
|
||
}
|
||
|
||
// getHotPostsByCursorFromDB 热门游标：优先 ZSET（一次多取 1 条判断 hasMore），否则回源 DB
|
||
func (s *postServiceImpl) getHotPostsByCursorFromDB(ctx context.Context, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Post], error) {
|
||
offset := 0
|
||
if req.Cursor != "" {
|
||
if parsed, err := strconv.Atoi(req.Cursor); err == nil && parsed >= 0 {
|
||
offset = parsed
|
||
}
|
||
}
|
||
|
||
if s.cache != nil {
|
||
if ids, hit := cache.GetTyped[[]string](s.cache, cache.HotRankTopIDsKey()); hit && len(ids) > 0 {
|
||
if offset < len(ids) {
|
||
stop := offset + req.PageSize + 1
|
||
if stop > len(ids) {
|
||
stop = len(ids)
|
||
}
|
||
slice := ids[offset:stop]
|
||
hasMore := len(slice) > req.PageSize
|
||
if hasMore {
|
||
slice = slice[:req.PageSize]
|
||
}
|
||
posts, gerr := s.postRepo.GetByIDs(slice)
|
||
if gerr != nil {
|
||
return nil, gerr
|
||
}
|
||
nextCursor := ""
|
||
if hasMore {
|
||
nextCursor = strconv.Itoa(offset + req.PageSize)
|
||
}
|
||
return &cursor.CursorPageResult[*model.Post]{
|
||
Items: posts,
|
||
NextCursor: nextCursor,
|
||
PrevCursor: "",
|
||
HasMore: hasMore,
|
||
}, nil
|
||
}
|
||
return &cursor.CursorPageResult[*model.Post]{
|
||
Items: []*model.Post{},
|
||
NextCursor: "",
|
||
PrevCursor: "",
|
||
HasMore: false,
|
||
}, nil
|
||
}
|
||
|
||
key := cache.HotRankZSetKey()
|
||
card, zerr := s.cache.ZCard(ctx, key)
|
||
if zerr == nil && card > 0 {
|
||
stop := int64(offset + req.PageSize)
|
||
ids, rerr := s.cache.ZRevRange(ctx, key, int64(offset), stop)
|
||
if rerr == nil && len(ids) > 0 {
|
||
hasMore := len(ids) > req.PageSize
|
||
if hasMore {
|
||
ids = ids[:req.PageSize]
|
||
}
|
||
posts, gerr := s.postRepo.GetByIDs(ids)
|
||
if gerr != nil {
|
||
return nil, gerr
|
||
}
|
||
nextCursor := ""
|
||
if hasMore {
|
||
nextCursor = strconv.Itoa(offset + req.PageSize)
|
||
}
|
||
return &cursor.CursorPageResult[*model.Post]{
|
||
Items: posts,
|
||
NextCursor: nextCursor,
|
||
PrevCursor: "",
|
||
HasMore: hasMore,
|
||
}, nil
|
||
}
|
||
}
|
||
}
|
||
|
||
page := offset/req.PageSize + 1
|
||
posts, _, err := s.postRepo.GetHotPosts(page, req.PageSize+1)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
hasMore := len(posts) > req.PageSize
|
||
if hasMore {
|
||
posts = posts[:req.PageSize]
|
||
}
|
||
|
||
nextCursor := ""
|
||
if hasMore {
|
||
nextCursor = strconv.Itoa(offset + req.PageSize)
|
||
}
|
||
|
||
return &cursor.CursorPageResult[*model.Post]{
|
||
Items: posts,
|
||
NextCursor: nextCursor,
|
||
PrevCursor: "",
|
||
HasMore: hasMore,
|
||
}, nil
|
||
}
|