From 631cb2203f74ea70b634630b3899bf2cdaf3f08c Mon Sep 17 00:00:00 2001 From: Leon <147289645+LeoninCS@users.noreply.github.com> Date: Tue, 30 Dec 2025 01:25:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AF=84=E8=AE=BA=E7=9B=B8=E5=85=B3rab?= =?UTF-8?q?bitMQ?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/middleware/rabbitmq/commentMQ.go | 71 ++++++++++ backend/internal/video/comment_handler.go | 9 +- backend/internal/video/comment_service.go | 71 +++++++++- backend/internal/worker/commentworker.go | 122 ++++++++++++++++++ 4 files changed, 263 insertions(+), 10 deletions(-) create mode 100644 backend/internal/middleware/rabbitmq/commentMQ.go create mode 100644 backend/internal/worker/commentworker.go diff --git a/backend/internal/middleware/rabbitmq/commentMQ.go b/backend/internal/middleware/rabbitmq/commentMQ.go new file mode 100644 index 0000000..c1cf0b4 --- /dev/null +++ b/backend/internal/middleware/rabbitmq/commentMQ.go @@ -0,0 +1,71 @@ +package rabbitmq + +import ( + "context" + "errors" + "time" +) + +type CommentMQ struct { + *RabbitMQ +} + +const ( + commentExchange = "comment.events" + commentQueue = "comment.events" + commentBindingKey = "comment.*" + + commentPublishRK = "comment.publish" + commentDeleteRK = "comment.delete" +) + +type CommentEvent struct { + EventID string `json:"event_id"` + Action string `json:"action"` + CommentID uint `json:"comment_id,omitempty"` + Username string `json:"username,omitempty"` + VideoID uint `json:"video_id,omitempty"` + AuthorID uint `json:"author_id,omitempty"` + Content string `json:"content,omitempty"` + OccurredAt time.Time `json:"occurred_at"` +} + +func NewCommentMQ(base *RabbitMQ) (*CommentMQ, error) { + if base == nil { + return nil, errors.New("rabbitmq base is nil") + } + if err := base.DeclareTopic(commentExchange, commentQueue, commentBindingKey); err != nil { + return nil, err + } + return &CommentMQ{RabbitMQ: base}, nil +} + +func (c *CommentMQ) Publish(ctx context.Context, username string, videoID, authorID uint, content string) error { + return c.publish(ctx, "publish", commentPublishRK, CommentEvent{ + Username: username, + VideoID: videoID, + AuthorID: authorID, + Content: content, + }) +} + +func (c *CommentMQ) Delete(ctx context.Context, commentID uint) error { + return c.publish(ctx, "delete", commentDeleteRK, CommentEvent{ + CommentID: commentID, + }) +} + +func (c *CommentMQ) publish(ctx context.Context, action, routingKey string, evt CommentEvent) error { + if c == nil || c.RabbitMQ == nil { + return errors.New("comment mq is not initialized") + } + id, err := newEventID(16) + if err != nil { + return err + } + evt.EventID = id + evt.Action = action + evt.OccurredAt = time.Now().UTC() + return c.PublishJSON(ctx, commentExchange, routingKey, evt) +} + diff --git a/backend/internal/video/comment_handler.go b/backend/internal/video/comment_handler.go index 7028796..021ae9a 100644 --- a/backend/internal/video/comment_handler.go +++ b/backend/internal/video/comment_handler.go @@ -10,11 +10,10 @@ import ( type CommentHandler struct { service *CommentService accountService *account.AccountService - videoService *VideoService } -func NewCommentHandler(service *CommentService, accountService *account.AccountService, videoService *VideoService) *CommentHandler { - return &CommentHandler{service: service, accountService: accountService, videoService: videoService} +func NewCommentHandler(service *CommentService, accountService *account.AccountService) *CommentHandler { + return &CommentHandler{service: service, accountService: accountService} } func (h *CommentHandler) PublishComment(c *gin.Context) { var req PublishCommentRequest @@ -50,10 +49,6 @@ func (h *CommentHandler) PublishComment(c *gin.Context) { c.JSON(400, gin.H{"error": err.Error()}) return } - if err := h.videoService.UpdatePopularity(c.Request.Context(), req.VideoID, 1); err != nil { - c.JSON(500, gin.H{"error": err.Error()}) - return - } c.JSON(200, gin.H{"message": "comment published successfully"}) } diff --git a/backend/internal/video/comment_service.go b/backend/internal/video/comment_service.go index 4d89b06..875c2b0 100644 --- a/backend/internal/video/comment_service.go +++ b/backend/internal/video/comment_service.go @@ -3,18 +3,38 @@ package video import ( "context" "errors" + "feedsystem_video_go/internal/middleware/rabbitmq" + rediscache "feedsystem_video_go/internal/middleware/redis" + "strings" + + "gorm.io/gorm" ) type CommentService struct { repo *CommentRepository VideoRepository *VideoRepository + cache *rediscache.Client + commentMQ *rabbitmq.CommentMQ + popularityMQ *rabbitmq.PopularityMQ } -func NewCommentService(repo *CommentRepository, videoRepo *VideoRepository) *CommentService { - return &CommentService{repo: repo, VideoRepository: videoRepo} +func NewCommentService(repo *CommentRepository, videoRepo *VideoRepository, cache *rediscache.Client, commentMQ *rabbitmq.CommentMQ, popularityMQ *rabbitmq.PopularityMQ) *CommentService { + return &CommentService{repo: repo, VideoRepository: videoRepo, cache: cache, commentMQ: commentMQ, popularityMQ: popularityMQ} } func (s *CommentService) Publish(ctx context.Context, comment *Comment) error { + if comment == nil { + return errors.New("comment is nil") + } + comment.Username = strings.TrimSpace(comment.Username) + comment.Content = strings.TrimSpace(comment.Content) + if comment.VideoID == 0 || comment.AuthorID == 0 { + return errors.New("video_id and author_id are required") + } + if comment.Content == "" { + return errors.New("content is required") + } + exists, err := s.VideoRepository.IsExist(ctx, comment.VideoID) if err != nil { return err @@ -22,7 +42,47 @@ func (s *CommentService) Publish(ctx context.Context, comment *Comment) error { if !exists { return errors.New("video not found") } - return s.repo.CreateComment(ctx, comment) + + mysqlEnqueued := false + redisEnqueued := false + if s.commentMQ != nil { + if err := s.commentMQ.Publish(ctx, comment.Username, comment.VideoID, comment.AuthorID, comment.Content); err == nil { + mysqlEnqueued = true + } + } + if s.popularityMQ != nil { + if err := s.popularityMQ.Update(ctx, comment.VideoID, 1); err == nil { + redisEnqueued = true + } + } + if mysqlEnqueued && redisEnqueued { + return nil + } + + // Fallback: direct MySQL write when comment MQ publish fails. + if !mysqlEnqueued { + if err := s.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Select("id").First(&Video{}, comment.VideoID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return errors.New("video not found") + } + return err + } + if err := tx.Create(comment).Error; err != nil { + return err + } + return tx.Model(&Video{}).Where("id = ?", comment.VideoID). + UpdateColumn("popularity", gorm.Expr("popularity + 1")).Error + }); err != nil { + return err + } + } + + // Fallback: direct Redis update when popularity MQ publish fails. + if !redisEnqueued { + UpdatePopularityCache(ctx, s.cache, comment.VideoID, 1) + } + return nil } func (s *CommentService) Delete(ctx context.Context, commentID uint, accountID uint) error { @@ -36,6 +96,11 @@ func (s *CommentService) Delete(ctx context.Context, commentID uint, accountID u if comment.AuthorID != accountID { return errors.New("permission denied") } + if s.commentMQ != nil { + if err := s.commentMQ.Delete(ctx, commentID); err == nil { + return nil + } + } return s.repo.DeleteComment(ctx, comment) } diff --git a/backend/internal/worker/commentworker.go b/backend/internal/worker/commentworker.go new file mode 100644 index 0000000..6b941b3 --- /dev/null +++ b/backend/internal/worker/commentworker.go @@ -0,0 +1,122 @@ +package worker + +import ( + "context" + "encoding/json" + "errors" + "feedsystem_video_go/internal/middleware/rabbitmq" + "feedsystem_video_go/internal/video" + "log" + "strings" + + amqp "github.com/rabbitmq/amqp091-go" +) + +type CommentWorker struct { + ch *amqp.Channel + comments *video.CommentRepository + videos *video.VideoRepository + queue string +} + +func NewCommentWorker(ch *amqp.Channel, comments *video.CommentRepository, videos *video.VideoRepository, queue string) *CommentWorker { + return &CommentWorker{ch: ch, comments: comments, videos: videos, queue: queue} +} + +func (w *CommentWorker) Run(ctx context.Context) error { + if w == nil || w.ch == nil || w.comments == nil || w.videos == nil { + return errors.New("comment worker is not initialized") + } + if w.queue == "" { + return errors.New("queue is required") + } + + deliveries, err := w.ch.Consume( + w.queue, + "", + false, + false, + false, + false, + nil, + ) + if err != nil { + return err + } + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case d, ok := <-deliveries: + if !ok { + return errors.New("deliveries channel closed") + } + w.handleDelivery(ctx, d) + } + } +} + +func (w *CommentWorker) handleDelivery(ctx context.Context, d amqp.Delivery) { + if err := w.process(ctx, d.Body); err != nil { + log.Printf("comment worker: failed to process message: %v", err) + _ = d.Nack(false, true) + return + } + _ = d.Ack(false) +} + +func (w *CommentWorker) process(ctx context.Context, body []byte) error { + var evt rabbitmq.CommentEvent + if err := json.Unmarshal(body, &evt); err != nil { + return nil + } + switch evt.Action { + case "publish": + return w.applyPublish(ctx, &evt) + case "delete": + return w.applyDelete(ctx, &evt) + default: + return nil + } +} + +func (w *CommentWorker) applyPublish(ctx context.Context, evt *rabbitmq.CommentEvent) error { + if evt == nil || evt.VideoID == 0 || evt.AuthorID == 0 || strings.TrimSpace(evt.Content) == "" { + return nil + } + + ok, err := w.videos.IsExist(ctx, evt.VideoID) + if err != nil { + return err + } + if !ok { + return nil + } + + c := &video.Comment{ + Username: strings.TrimSpace(evt.Username), + VideoID: evt.VideoID, + AuthorID: evt.AuthorID, + Content: strings.TrimSpace(evt.Content), + } + if err := w.comments.CreateComment(ctx, c); err != nil { + return err + } + return w.videos.ChangePopularity(ctx, evt.VideoID, 1) +} + +func (w *CommentWorker) applyDelete(ctx context.Context, evt *rabbitmq.CommentEvent) error { + if evt == nil || evt.CommentID == 0 { + return nil + } + c, err := w.comments.GetByID(ctx, evt.CommentID) + if err != nil { + return err + } + if c == nil { + return nil + } + return w.comments.DeleteComment(ctx, c) +} +