d8e56e60dc
Add client_request_id field to post creation endpoints (CreateVotePost and CreatePostWithSegments). When provided, the server uses Redis SetNX-based idempotency to prevent duplicate posts from double-clicks or network retries within a 24-hour TTL. Duplicate requests return the originally created post; concurrent in-flight requests return 429 "正在发布中，请稍候".
438 lines
13 KiB
Go
438 lines
13 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
|
||
"with_you/internal/cache"
|
||
"with_you/internal/dto"
|
||
apperrors "with_you/internal/errors"
|
||
"with_you/internal/model"
|
||
"with_you/internal/pkg/hook"
|
||
"with_you/internal/repository"
|
||
)
|
||
|
||
// VoteService 投票服务接口
|
||
type VoteService interface {
|
||
CreateVotePost(ctx context.Context, userID string, req *dto.CreateVotePostRequest) (*dto.PostResponse, error)
|
||
GetVoteOptions(postID string) ([]dto.VoteOptionDTO, error)
|
||
GetVoteResult(postID, userID string) (*dto.VoteResultDTO, error)
|
||
Vote(ctx context.Context, postID, userID, optionID string) error
|
||
Unvote(ctx context.Context, postID, userID string) error
|
||
UpdateVoteOption(ctx context.Context, postID, optionID, userID, content string) error
|
||
}
|
||
|
||
// voteService 投票服务实现
|
||
type voteService struct {
|
||
voteRepo repository.VoteRepository
|
||
postRepo repository.PostRepository
|
||
cache cache.Cache
|
||
postAIService PostAIService
|
||
systemMessageService SystemMessageService
|
||
hookManager *hook.Manager
|
||
logService *LogService
|
||
}
|
||
|
||
func NewVoteService(
|
||
voteRepo repository.VoteRepository,
|
||
postRepo repository.PostRepository,
|
||
cache cache.Cache,
|
||
postAIService PostAIService,
|
||
systemMessageService SystemMessageService,
|
||
hookManager *hook.Manager,
|
||
logService *LogService,
|
||
) VoteService {
|
||
return &voteService{
|
||
voteRepo: voteRepo,
|
||
postRepo: postRepo,
|
||
cache: cache,
|
||
postAIService: postAIService,
|
||
systemMessageService: systemMessageService,
|
||
hookManager: hookManager,
|
||
logService: logService,
|
||
}
|
||
}
|
||
|
||
// CreateVotePost 创建投票帖子
|
||
func (s *voteService) CreateVotePost(ctx context.Context, userID string, req *dto.CreateVotePostRequest) (*dto.PostResponse, error) {
|
||
if len(req.VoteOptions) < 2 {
|
||
return nil, errors.New("投票选项至少需要2个")
|
||
}
|
||
if len(req.VoteOptions) > 10 {
|
||
return nil, errors.New("投票选项最多10个")
|
||
}
|
||
|
||
// 幂等检查：同一 userID + clientRequestID 在 TTL 内只创建一次。
|
||
// 重复请求（多次点击、网络重试）会命中已回填的 postID，直接返回首次创建的帖子。
|
||
if req.ClientRequestID != "" && s.cache != nil {
|
||
idemKey := cache.PostCreateIdempotentKey(userID, req.ClientRequestID)
|
||
result := cache.TryAcquireIdempotency(ctx, s.cache, idemKey, PostCreateIdempotentTTL)
|
||
if !result.Acquired {
|
||
// 命中已完成的请求：返回首次创建的帖子。
|
||
if result.ExistingValue != "" {
|
||
if existing, err := s.postRepo.GetByID(result.ExistingValue); err == nil && existing != nil {
|
||
return s.convertToPostResponse(existing, userID), nil
|
||
}
|
||
}
|
||
// 命中"进行中"占位：前一个请求还在处理，拒绝重复提交。
|
||
return nil, ErrDuplicatePostRequest
|
||
}
|
||
}
|
||
|
||
// 构建 vote segment
|
||
voteOptions := make([]dto.VoteOptionData, len(req.VoteOptions))
|
||
for i, opt := range req.VoteOptions {
|
||
voteOptions[i] = dto.VoteOptionData{
|
||
ID: fmt.Sprintf("opt_%d", i+1),
|
||
Content: opt,
|
||
}
|
||
}
|
||
segments := model.MessageSegments{
|
||
dto.NewTextSegment(req.Content),
|
||
{
|
||
Type: string(dto.SegmentTypeVote),
|
||
Data: map[string]any{
|
||
"options": voteOptions,
|
||
"max_choices": 1,
|
||
"is_multi_choice": false,
|
||
},
|
||
},
|
||
}
|
||
|
||
content := req.Content
|
||
if content == "" {
|
||
content = " "
|
||
}
|
||
|
||
post := &model.Post{
|
||
UserID: userID,
|
||
ChannelID: req.ChannelID,
|
||
Title: req.Title,
|
||
Content: content,
|
||
Segments: segments,
|
||
Status: model.PostStatusPending,
|
||
IsVote: true,
|
||
}
|
||
|
||
err := s.postRepo.Create(post, req.Images)
|
||
if err != nil {
|
||
// 创建失败：释放幂等占位，允许用户重试。
|
||
if req.ClientRequestID != "" && s.cache != nil {
|
||
cache.ReleaseIdempotency(ctx, s.cache, cache.PostCreateIdempotentKey(userID, req.ClientRequestID))
|
||
}
|
||
return nil, err
|
||
}
|
||
|
||
// 创建成功：回填幂等键为 postID，使后续重复请求返回同一帖子。
|
||
if req.ClientRequestID != "" && s.cache != nil {
|
||
cache.CompleteIdempotency(ctx, s.cache, cache.PostCreateIdempotentKey(userID, req.ClientRequestID), post.ID, PostCreateIdempotentTTL)
|
||
}
|
||
|
||
// 异步审核
|
||
go s.reviewVotePostAsync(post.ID, userID, req.Title, content, req.Images)
|
||
|
||
// 重新查询以获取关联的User和Images
|
||
createdPost, err := s.postRepo.GetByID(post.ID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return s.convertToPostResponse(createdPost, userID), nil
|
||
}
|
||
|
||
func (s *voteService) reviewVotePostAsync(postID, userID, title, content string, images []string) {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[ERROR] Panic in vote 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 vote post %s after panic recovery: %v", postID, err)
|
||
}
|
||
}
|
||
}()
|
||
|
||
if s.hookManager == nil {
|
||
if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "system"); err != nil {
|
||
log.Printf("[WARN] Failed to publish vote post without hook manager: %v", err)
|
||
} else {
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
}
|
||
return
|
||
}
|
||
|
||
var authorID uint
|
||
if _, err := fmt.Sscanf(userID, "%d", &authorID); err != nil {
|
||
log.Printf("[WARN] Invalid userID format in vote moderation: %q, err: %v", userID, err)
|
||
if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "system"); err != nil {
|
||
log.Printf("[WARN] Failed to publish vote post %s after invalid userID: %v", postID, err)
|
||
} else {
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
}
|
||
return
|
||
}
|
||
|
||
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 {
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
return
|
||
}
|
||
if updateErr := s.updateModerationStatusWithRetry(postID, model.PostStatusRejected, result.RejectReason, result.ReviewedBy); updateErr != nil {
|
||
log.Printf("[WARN] Failed to reject vote post %s: %v", postID, updateErr)
|
||
} else {
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
}
|
||
s.notifyModerationRejected(userID, result.RejectReason)
|
||
return
|
||
}
|
||
|
||
if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", result.ReviewedBy); err != nil {
|
||
log.Printf("[WARN] Failed to publish vote post %s: %v", postID, err)
|
||
return
|
||
}
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
cache.InvalidatePostList(s.cache)
|
||
}
|
||
|
||
func (s *voteService) 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 for vote post 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 *voteService) 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() {
|
||
_ = s.systemMessageService.SendSystemAnnouncement(
|
||
context.Background(),
|
||
[]string{userID},
|
||
"投票帖审核未通过",
|
||
content,
|
||
)
|
||
}()
|
||
}
|
||
|
||
// resolveVoteOptions 解析投票选项（优先从 segments 读取，兼容旧数据从 vote_options 表读取）
|
||
func (s *voteService) resolveVoteOptions(postID string, segments model.MessageSegments) ([]dto.VoteOptionDTO, error) {
|
||
if len(segments) > 0 {
|
||
voteData := dto.ExtractVoteSegmentData(segments)
|
||
if voteData != nil {
|
||
result := make([]dto.VoteOptionDTO, 0, len(voteData.Options))
|
||
for _, opt := range voteData.Options {
|
||
count, err := s.voteRepo.CountVotesByOption(postID, opt.ID)
|
||
if err != nil {
|
||
log.Printf("[WARN] Failed to count votes for option %s in post %s: %v", opt.ID, postID, err)
|
||
}
|
||
result = append(result, dto.VoteOptionDTO{
|
||
ID: opt.ID,
|
||
Content: opt.Content,
|
||
VotesCount: count,
|
||
})
|
||
}
|
||
return result, nil
|
||
}
|
||
}
|
||
|
||
options, err := s.voteRepo.GetOptionsByPostID(postID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
result := make([]dto.VoteOptionDTO, 0, len(options))
|
||
for _, option := range options {
|
||
result = append(result, dto.VoteOptionDTO{
|
||
ID: option.ID,
|
||
Content: option.Content,
|
||
VotesCount: option.VotesCount,
|
||
})
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// GetVoteOptions 获取投票选项（优先从 segments 读取，兼容旧数据从 vote_options 表读取）
|
||
func (s *voteService) GetVoteOptions(postID string) ([]dto.VoteOptionDTO, error) {
|
||
post, err := s.postRepo.GetByID(postID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return s.resolveVoteOptions(postID, post.Segments)
|
||
}
|
||
|
||
// GetVoteResult 获取投票结果（包含用户投票状态）
|
||
func (s *voteService) GetVoteResult(postID, userID string) (*dto.VoteResultDTO, error) {
|
||
post, err := s.postRepo.GetByID(postID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
optionDTOs, err := s.resolveVoteOptions(postID, post.Segments)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
userVote, err := s.voteRepo.GetUserVote(postID, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
result := &dto.VoteResultDTO{
|
||
Options: optionDTOs,
|
||
TotalVotes: 0,
|
||
HasVoted: userVote != nil,
|
||
}
|
||
|
||
if userVote != nil {
|
||
result.VotedOptionID = userVote.OptionID
|
||
}
|
||
|
||
for _, opt := range optionDTOs {
|
||
result.TotalVotes += opt.VotesCount
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// Vote 投票
|
||
func (s *voteService) Vote(ctx context.Context, postID, userID, optionID string) error {
|
||
// 调用voteRepo.Vote
|
||
err := s.voteRepo.Vote(postID, userID, optionID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
|
||
return nil
|
||
}
|
||
|
||
// Unvote 取消投票
|
||
func (s *voteService) Unvote(ctx context.Context, postID, userID string) error {
|
||
// 调用voteRepo.Unvote
|
||
err := s.voteRepo.Unvote(postID, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效帖子详情缓存
|
||
cache.InvalidatePostDetail(s.cache, postID)
|
||
|
||
return nil
|
||
}
|
||
|
||
// UpdateVoteOption 更新投票选项（作者权限）
|
||
func (s *voteService) UpdateVoteOption(ctx context.Context, postID, optionID, userID, content string) error {
|
||
// 获取帖子信息
|
||
post, err := s.postRepo.GetByID(postID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 验证用户是否为帖子作者
|
||
if post.UserID != userID {
|
||
return apperrors.ErrForbidden
|
||
}
|
||
|
||
// 调用voteRepo.UpdateOption
|
||
return s.voteRepo.UpdateOption(optionID, content)
|
||
}
|
||
|
||
// convertToPostResponse 将Post模型转换为PostResponse DTO
|
||
func (s *voteService) convertToPostResponse(post *model.Post, currentUserID string) *dto.PostResponse {
|
||
if post == nil {
|
||
return nil
|
||
}
|
||
|
||
response := &dto.PostResponse{
|
||
ID: post.ID,
|
||
UserID: post.UserID,
|
||
Title: post.Title,
|
||
Content: post.Content,
|
||
Segments: dto.SegmentsOrDefault(post.Segments, post.Content),
|
||
LikesCount: post.LikesCount,
|
||
CommentsCount: post.CommentsCount,
|
||
FavoritesCount: post.FavoritesCount,
|
||
SharesCount: post.SharesCount,
|
||
ViewsCount: post.ViewsCount,
|
||
IsPinned: post.IsPinned,
|
||
IsLocked: post.IsLocked,
|
||
IsVote: post.IsVote,
|
||
CreatedAt: dto.FormatTime(post.CreatedAt),
|
||
UpdatedAt: dto.FormatTime(post.UpdatedAt),
|
||
ContentEditedAt: dto.FormatTimePointer(post.ContentEditedAt),
|
||
Images: make([]dto.PostImageResponse, 0, len(post.Images)),
|
||
}
|
||
|
||
// 转换图片
|
||
for _, img := range post.Images {
|
||
response.Images = append(response.Images, dto.PostImageResponse{
|
||
ID: img.ID,
|
||
URL: img.URL,
|
||
ThumbnailURL: img.ThumbnailURL,
|
||
Width: img.Width,
|
||
Height: img.Height,
|
||
})
|
||
}
|
||
|
||
// 转换作者信息
|
||
if post.User != nil {
|
||
response.Author = &dto.UserResponse{
|
||
ID: post.User.ID,
|
||
Username: post.User.Username,
|
||
Nickname: post.User.Nickname,
|
||
Avatar: post.User.Avatar,
|
||
}
|
||
}
|
||
|
||
return response
|
||
}
|