diff --git a/internal/cache/idempotency.go b/internal/cache/idempotency.go new file mode 100644 index 0000000..8048b54 --- /dev/null +++ b/internal/cache/idempotency.go @@ -0,0 +1,122 @@ +package cache + +import ( + "context" + "time" + + redislib "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// IdempotencyResult 描述幂等检查的结果。 +type IdempotencyResult struct { + // Acquired 为 true 表示当前请求成功抢占到幂等键，调用方应执行业务逻辑。 + // 为 false 表示检测到重复请求：ExistingValue 为先前写入的值（可能为空， + // 表示前一个请求仍在进行中——即"进行中"占位）。 + Acquired bool + // ExistingValue 命中已有幂等记录时的值（如已创建的 postID）。 + // Acquired=true 时为空字符串。 + ExistingValue string +} + +// TryAcquireIdempotency 原子抢占幂等键。 +// +// 流程： +// 1. 读取幂等键。若已有值，视为重复请求，返回 (Acquired=false, ExistingValue=...)。 +// 2. 若无值，用 SetNX 写入空占位（标记"进行中"）。成功则获得执行权 (Acquired=true)。 +// +// 降级：当 Redis 不可用（cache 非 *LayeredCache/*RedisCache，或底层 client 为 nil）时， +// 返回 (Acquired=true)，即不阻断业务（与现有 call_service 的降级策略一致）， +// 此时幂等性退化为依赖前端防抖。 +// +// 调用方在业务成功后必须调用 CompleteIdempotency 写入真实结果，以便后续重试可返回同一资源。 +func TryAcquireIdempotency(ctx context.Context, c Cache, key string, ttl time.Duration) IdempotencyResult { + if c == nil { + return IdempotencyResult{Acquired: true} + } + + // 1) 先查命中：若已有"完成值"，直接复用，避免重复创建。 + if existing, ok := GetTyped[string](c, key); ok && existing != "" { + return IdempotencyResult{Acquired: false, ExistingValue: existing} + } + + // 2) 用 SetNX 写入空占位，原子抢占执行权。 + // 占位为空字符串：表示"进行中"。后续命中时若值为空，视为并发请求进行中， + // 直接拒绝（Acquired=false 且 ExistingValue=""）。 + rdb := resolveRedisClient(c) + if rdb == nil { + // Redis 不可用：降级放行，依赖前端防抖。 + return IdempotencyResult{Acquired: true} + } + + ok, err := rdb.SetNX(ctx, normalizeKey(key), "", ttl).Result() + if err != nil { + zap.L().Warn("idempotency SetNX failed, falling back to non-idempotent create", + zap.String("key", key), + zap.Error(err), + ) + return IdempotencyResult{Acquired: true} + } + if !ok { + // 已被其他请求抢占（进行中或已完成但值尚未回填）。 + // 读取实际值以区分"进行中"与"已完成"。 + if existing, ok := GetTyped[string](c, key); ok && existing != "" { + return IdempotencyResult{Acquired: false, ExistingValue: existing} + } + return IdempotencyResult{Acquired: false} + } + return IdempotencyResult{Acquired: true} +} + +// CompleteIdempotency 在业务成功后回填幂等键的真实结果值（如 postID）， +// 使后续重复请求能返回同一资源。失败时仅记录日志，不影响业务结果。 +func CompleteIdempotency(ctx context.Context, c Cache, key, value string, ttl time.Duration) { + if c == nil || value == "" { + return + } + // 直接覆盖占位为真实值，保留原 TTL。 + rdb := resolveRedisClient(c) + if rdb == nil { + // 降级：用通用 Set 写入（LayeredCache/RedisCache 都支持）。 + c.Set(key, value, ttl) + return + } + if err := rdb.Set(ctx, normalizeKey(key), value, ttl).Err(); err != nil { + zap.L().Warn("idempotency complete Set failed", + zap.String("key", key), + zap.Error(err), + ) + } +} + +// ReleaseIdempotency 在业务失败时清除占位，允许重试。 +func ReleaseIdempotency(ctx context.Context, c Cache, key string) { + if c == nil { + return + } + // 仅当当前值仍为空占位时才删除，避免误删已回填的真实值。 + if existing, ok := GetTyped[string](c, key); ok && existing != "" { + return + } + c.Delete(key) +} + +// resolveRedisClient 从 Cache 实现中解析出底层 *redis.Client，用于 SetNX 原子操作。 +// 返回 nil 表示 Redis 不可用（如纯内存 LRU 模式）。 +func resolveRedisClient(c Cache) *redislib.Client { + switch v := c.(type) { + case *LayeredCache: + client := v.GetRedisClient() + if client == nil { + return nil + } + return client.GetClient() + case *RedisCache: + client := v.GetRedisClient() + if client == nil { + return nil + } + return client.GetClient() + } + return nil +} diff --git a/internal/cache/keys.go b/internal/cache/keys.go index 996b5c4..c6af7ad 100644 --- a/internal/cache/keys.go +++ b/internal/cache/keys.go @@ -38,12 +38,15 @@ const ( PrefixChannelsAllList = "channels:all_list" // 消息缓存相关 - keyPrefixMsgHash = "msg_hash" // 消息详情 Hash - keyPrefixMsgIndex = "msg_index" // 消息索引 Sorted Set - keyPrefixMsgCount = "msg_count" // 消息计数 - keyPrefixMsgSeq = "msg_seq" // Seq 计数器 - keyPrefixMsgPage = "msg_page" // 分页缓存 - keyPrefixMsgIdempotent = "msg_idem" // 消息幂等键 + keyPrefixMsgHash = "msg_hash" // 消息详情 Hash + keyPrefixMsgIndex = "msg_index" // 消息索引 Sorted Set + keyPrefixMsgCount = "msg_count" // 消息计数 + keyPrefixMsgSeq = "msg_seq" // Seq 计数器 + keyPrefixMsgPage = "msg_page" // 分页缓存 + keyPrefixMsgIdempotent = "msg_idem" // 消息幂等键 + + // 帖子创建幂等键前缀（clientRequestID -> 已创建的 postID） + keyPrefixPostIdempotent = "post_idem" ) // PostListKey 生成帖子列表缓存键 @@ -197,6 +200,12 @@ func MessageIdempotentKey(senderID, clientMsgID string) string { return fmt.Sprintf("%s:%s:%s", keyPrefixMsgIdempotent, senderID, clientMsgID) } +// PostCreateIdempotentKey 帖子创建幂等键 (userID + clientRequestID -> postID) +// 用于在 SetNX 原子抢占期间识别同一客户端请求的重试，避免重复发帖。 +func PostCreateIdempotentKey(userID, clientRequestID string) string { + return fmt.Sprintf("%s:%s:%s", keyPrefixPostIdempotent, userID, clientRequestID) +} + // UserReadSeqKey 用户已读位置缓存键（OpenIM 风格 SEQ_USER_READ:{convID}:{userID}） func UserReadSeqKey(convID, userID string) string { return fmt.Sprintf("%s:%s:%s", PrefixUserReadSeq, convID, userID) diff --git a/internal/dto/post_dto.go b/internal/dto/post_dto.go index b659334..e276055 100644 --- a/internal/dto/post_dto.go +++ b/internal/dto/post_dto.go @@ -54,20 +54,26 @@ type PostResponse struct { // CreateVotePostRequest 创建投票帖子请求 type CreateVotePostRequest struct { - Title string `json:"title" binding:"required,max=200"` - Content string `json:"content" binding:"max=2000"` - ChannelID *string `json:"channel_id,omitempty"` - Images []string `json:"images"` - VoteOptions []string `json:"vote_options" binding:"required,min=2,max=10"` + Title string `json:"title" binding:"required,max=200"` + Content string `json:"content" binding:"max=2000"` + ChannelID *string `json:"channel_id,omitempty"` + Images []string `json:"images"` + VoteOptions []string `json:"vote_options" binding:"required,min=2,max=10"` + // ClientRequestID 客户端为本次发帖生成的幂等键（UUID）。 + // 同一 userID + ClientRequestID 在 TTL 内重复提交时，后端直接返回首次创建的帖子，避免重复发帖。 + ClientRequestID string `json:"client_request_id,omitempty"` } // CreatePostWithSegmentsRequest 创建帖子请求（含 segments） type CreatePostWithSegmentsRequest struct { - Title string `json:"title" binding:"required,max=200"` - Content string `json:"content"` - Segments model.MessageSegments `json:"segments"` - Images []string `json:"images"` - ChannelID *string `json:"channel_id,omitempty"` + Title string `json:"title" binding:"required,max=200"` + Content string `json:"content"` + Segments model.MessageSegments `json:"segments"` + Images []string `json:"images"` + ChannelID *string `json:"channel_id,omitempty"` + // ClientRequestID 客户端为本次发帖生成的幂等键（UUID）。 + // 同一 userID + ClientRequestID 在 TTL 内重复提交时，后端直接返回首次创建的帖子，避免重复发帖。 + ClientRequestID string `json:"client_request_id,omitempty"` } // VoteOptionDTO 投票选项DTO diff --git a/internal/errors/app_errors.go b/internal/errors/app_errors.go index bea87e8..207241a 100644 --- a/internal/errors/app_errors.go +++ b/internal/errors/app_errors.go @@ -130,6 +130,9 @@ var ( // 学习资料相关错误 ErrSubjectHasMaterials = &AppError{Code: "SUBJECT_HAS_MATERIALS", Message: "学科下存在资料，无法删除"} + // 帖子创建幂等性：同一 client_request_id 的请求仍在处理中（占位未回填真实结果） + ErrDuplicatePostRequest = &AppError{Code: "DUPLICATE_POST_REQUEST", Message: "正在发布中，请稍候"} + // 身份认证相关错误 ErrVerificationPending = &AppError{Code: "VERIFICATION_PENDING", Message: "已有待审核的认证申请，请等待审核"} ErrVerificationNotFound = &AppError{Code: "VERIFICATION_NOT_FOUND", Message: "认证记录不存在"} diff --git a/internal/handler/post_handler.go b/internal/handler/post_handler.go index 953fb7b..2401879 100644 --- a/internal/handler/post_handler.go +++ b/internal/handler/post_handler.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log" + "net/http" "strconv" "github.com/gin-gonic/gin" @@ -105,13 +106,18 @@ func (h *PostHandler) Create(c *gin.Context) { return } - post, err := h.postService.Create(c.Request.Context(), userID, req.Title, content, segments, req.Images, req.ChannelID) + post, err := h.postService.Create(c.Request.Context(), userID, req.Title, content, segments, req.Images, req.ChannelID, req.ClientRequestID) if err != nil { var moderationErr *service.PostModerationRejectedError if errors.As(err, &moderationErr) { response.BadRequest(c, moderationErr.UserMessage()) return } + // 幂等命中"进行中"占位：提示客户端稍候，避免重复提交 + if errors.Is(err, service.ErrDuplicatePostRequest) { + response.ErrorWithStatusCode(c, http.StatusTooManyRequests, 429, err.Error()) + return + } response.InternalServerError(c, "failed to create post") return } diff --git a/internal/handler/vote_handler.go b/internal/handler/vote_handler.go index 8488196..c4fcb9a 100644 --- a/internal/handler/vote_handler.go +++ b/internal/handler/vote_handler.go @@ -47,6 +47,11 @@ func (h *VoteHandler) CreateVotePost(c *gin.Context) { response.BadRequest(c, moderationErr.UserMessage()) return } + // 幂等命中"进行中"占位：提示客户端稍候，避免重复提交 + if errors.Is(err, service.ErrDuplicatePostRequest) { + response.ErrorWithStatusCode(c, http.StatusTooManyRequests, 429, err.Error()) + return + } response.Error(c, http.StatusBadRequest, err.Error()) return } diff --git a/internal/service/post_service.go b/internal/service/post_service.go index 600d4c9..316be5f 100644 --- a/internal/service/post_service.go +++ b/internal/service/post_service.go @@ -22,12 +22,17 @@ const ( PostListTTL = 30 * time.Second // 帖子列表缓存30秒 PostListNullTTL = 5 * time.Second PostListJitterRatio = 0.15 + + // PostCreateIdempotentTTL 帖子创建幂等键 TTL：24 小时内同一 client_request_id 不会重复发帖 + PostCreateIdempotentTTL = 24 * time.Hour ) // PostService 帖子服务接口 type PostService interface { // 帖子CRUD - Create(ctx context.Context, userID, title, content string, segments model.MessageSegments, images []string, channelID *string) (*model.Post, error) + // Create 创建帖子。clientRequestID 为客户端幂等键（可空）： + // 非空时，同一 userID+clientRequestID 在 TTL 内重复提交将返回首次创建的帖子。 + Create(ctx context.Context, userID, title, content string, segments model.MessageSegments, images []string, channelID *string, clientRequestID 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 @@ -99,8 +104,29 @@ type PostListResult struct { Total int64 } +// ErrDuplicatePostRequest 表示检测到重复的发帖请求（幂等命中"进行中"占位）。 +// 客户端遇到此错误时应提示"正在发布中，请稍候"，而非重复提交。 +var ErrDuplicatePostRequest = apperrors.ErrDuplicatePostRequest + // Create 创建帖子 -func (s *postServiceImpl) Create(ctx context.Context, userID, title, content string, segments model.MessageSegments, images []string, channelID *string) (*model.Post, error) { +func (s *postServiceImpl) Create(ctx context.Context, userID, title, content string, segments model.MessageSegments, images []string, channelID *string, clientRequestID string) (*model.Post, error) { + // 幂等检查：同一 userID + clientRequestID 在 TTL 内只创建一次。 + // 重复请求（多次点击、网络重试）会命中已回填的 postID，直接返回首次创建的帖子。 + if clientRequestID != "" && s.cache != nil { + idemKey := cache.PostCreateIdempotentKey(userID, 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 existing, nil + } + } + // 命中"进行中"占位：前一个请求还在处理，拒绝重复提交。 + return nil, ErrDuplicatePostRequest + } + } + // 如果没有提供 content 但有 segments，从 segments 提取文本摘要 if content == "" && len(segments) > 0 { content = dto.ExtractTextContent(segments) @@ -125,9 +151,18 @@ func (s *postServiceImpl) Create(ctx context.Context, userID, title, content str err := s.postRepo.Create(post, images) if err != nil { + // 创建失败：释放幂等占位，允许用户重试。 + if clientRequestID != "" && s.cache != nil { + cache.ReleaseIdempotency(ctx, s.cache, cache.PostCreateIdempotentKey(userID, clientRequestID)) + } return nil, err } + // 创建成功：回填幂等键为 postID，使后续重复请求返回同一帖子。 + if clientRequestID != "" && s.cache != nil { + cache.CompleteIdempotency(ctx, s.cache, cache.PostCreateIdempotentKey(userID, clientRequestID), post.ID, PostCreateIdempotentTTL) + } + // 记录操作日志 if s.logService != nil { s.logService.OperationLog.RecordOperation(&model.OperationLog{ diff --git a/internal/service/vote_service.go b/internal/service/vote_service.go index 41e7dc7..ef6da16 100644 --- a/internal/service/vote_service.go +++ b/internal/service/vote_service.go @@ -66,6 +66,23 @@ func (s *voteService) CreateVotePost(ctx context.Context, userID string, req *dt 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 { @@ -103,9 +120,18 @@ func (s *voteService) CreateVotePost(ctx context.Context, userID string, req *dt 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)