From 38003731661be7ffed279a7bd91ac291e8acc548 Mon Sep 17 00:00:00 2001 From: Leon <147289645+LeoninCS@users.noreply.github.com> Date: Tue, 30 Dec 2025 01:23:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=82=B9=E8=B5=9E=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/likeMQ.go | 67 +++++++ backend/internal/video/like_handler.go | 15 +- backend/internal/video/like_repo.go | 27 +++ backend/internal/video/like_service.go | 178 +++++++++++++++--- backend/internal/worker/likeworker.go | 136 +++++++++++++ 5 files changed, 380 insertions(+), 43 deletions(-) create mode 100644 backend/internal/middleware/rabbitmq/likeMQ.go create mode 100644 backend/internal/worker/likeworker.go diff --git a/backend/internal/middleware/rabbitmq/likeMQ.go b/backend/internal/middleware/rabbitmq/likeMQ.go new file mode 100644 index 0000000..f9b1d30 --- /dev/null +++ b/backend/internal/middleware/rabbitmq/likeMQ.go @@ -0,0 +1,67 @@ +package rabbitmq + +import ( + "context" + "errors" + "time" +) + +type LikeMQ struct { + *RabbitMQ +} + +const ( + likeExchange = "like.events" + likeQueue = "like.events" + likeBindingKey = "like.*" + + likeLikeRK = "like.like" + likeUnlikeRK = "like.unlike" +) + +type LikeEvent struct { + EventID string `json:"event_id"` + Action string `json:"action"` + UserID uint `json:"user_id"` + VideoID uint `json:"video_id"` + OccurredAt time.Time `json:"occurred_at"` +} + +func NewLikeMQ(base *RabbitMQ) (*LikeMQ, error) { + if base == nil { + return nil, errors.New("rabbitmq base is nil") + } + if err := base.DeclareTopic(likeExchange, likeQueue, likeBindingKey); err != nil { + return nil, err + } + return &LikeMQ{RabbitMQ: base}, nil +} + +func (l *LikeMQ) Like(ctx context.Context, userID, videoID uint) error { + return l.publish(ctx, "like", likeLikeRK, userID, videoID) +} + +func (l *LikeMQ) Unlike(ctx context.Context, userID, videoID uint) error { + return l.publish(ctx, "unlike", likeUnlikeRK, userID, videoID) +} + +func (l *LikeMQ) publish(ctx context.Context, action, routingKey string, userID, videoID uint) error { + if l == nil || l.RabbitMQ == nil { + return errors.New("like mq is not initialized") + } + if userID == 0 || videoID == 0 { + return errors.New("userID and videoID are required") + } + id, err := newEventID(16) + if err != nil { + return err + } + event := LikeEvent{ + EventID: id, + Action: action, + UserID: userID, + VideoID: videoID, + OccurredAt: time.Now(), + } + return l.PublishJSON(ctx, likeExchange, routingKey, event) +} diff --git a/backend/internal/video/like_handler.go b/backend/internal/video/like_handler.go index 407836f..c0c0b62 100644 --- a/backend/internal/video/like_handler.go +++ b/backend/internal/video/like_handler.go @@ -7,12 +7,11 @@ import ( ) type LikeHandler struct { - service *LikeService - videoService *VideoService + service *LikeService } -func NewLikeHandler(service *LikeService, videoService *VideoService) *LikeHandler { - return &LikeHandler{service: service, videoService: videoService} +func NewLikeHandler(service *LikeService) *LikeHandler { + return &LikeHandler{service: service} } func (lh *LikeHandler) Like(c *gin.Context) { @@ -40,10 +39,6 @@ func (lh *LikeHandler) Like(c *gin.Context) { c.JSON(500, gin.H{"error": err.Error()}) return } - if err := lh.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": "like success"}) } @@ -72,10 +67,6 @@ func (lh *LikeHandler) Unlike(c *gin.Context) { c.JSON(500, gin.H{"error": err.Error()}) return } - if err := lh.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": "unlike success"}) } diff --git a/backend/internal/video/like_repo.go b/backend/internal/video/like_repo.go index 57936f3..28a72fe 100644 --- a/backend/internal/video/like_repo.go +++ b/backend/internal/video/like_repo.go @@ -2,7 +2,9 @@ package video import ( "context" + "errors" + "github.com/go-sql-driver/mysql" "gorm.io/gorm" ) @@ -24,6 +26,31 @@ func (r *LikeRepository) Unlike(ctx context.Context, like *Like) error { Delete(&Like{}).Error } +func (r *LikeRepository) LikeIgnoreDuplicate(ctx context.Context, like *Like) (created bool, err error) { + if like == nil || like.VideoID == 0 || like.AccountID == 0 { + return false, nil + } + err = r.db.WithContext(ctx).Create(like).Error + if err == nil { + return true, nil + } + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + return false, nil + } + return false, err +} + +func (r *LikeRepository) DeleteByVideoAndAccount(ctx context.Context, videoID, accountID uint) (deleted bool, err error) { + if videoID == 0 || accountID == 0 { + return false, nil + } + res := r.db.WithContext(ctx). + Where("video_id = ? AND account_id = ?", videoID, accountID). + Delete(&Like{}) + return res.RowsAffected > 0, res.Error +} + func (r *LikeRepository) IsLiked(ctx context.Context, videoID, accountID uint) (bool, error) { var count int64 err := r.db.WithContext(ctx).Model(&Like{}). diff --git a/backend/internal/video/like_service.go b/backend/internal/video/like_service.go index dfd0fe4..4196ed8 100644 --- a/backend/internal/video/like_service.go +++ b/backend/internal/video/like_service.go @@ -3,6 +3,8 @@ package video import ( "context" "errors" + "feedsystem_video_go/internal/middleware/rabbitmq" + rediscache "feedsystem_video_go/internal/middleware/redis" "time" "github.com/go-sql-driver/mysql" @@ -10,12 +12,15 @@ import ( ) type LikeService struct { - repo *LikeRepository - VideoRepo *VideoRepository + repo *LikeRepository + VideoRepo *VideoRepository + cache *rediscache.Client + likeMQ *rabbitmq.LikeMQ + popularityMQ *rabbitmq.PopularityMQ } -func NewLikeService(repo *LikeRepository, videoRepo *VideoRepository) *LikeService { - return &LikeService{repo: repo, VideoRepo: videoRepo} +func NewLikeService(repo *LikeRepository, videoRepo *VideoRepository, cache *rediscache.Client, likeMQ *rabbitmq.LikeMQ, popularityMQ *rabbitmq.PopularityMQ) *LikeService { + return &LikeService{repo: repo, VideoRepo: videoRepo, cache: cache, likeMQ: likeMQ, popularityMQ: popularityMQ} } func isDupKey(err error) bool { @@ -24,41 +29,152 @@ func isDupKey(err error) bool { } func (s *LikeService) Like(ctx context.Context, like *Like) error { + if like == nil { + return errors.New("like is nil") + } + if like.VideoID == 0 || like.AccountID == 0 { + return errors.New("video_id and account_id are required") + } + + if s.VideoRepo != nil { + ok, err := s.VideoRepo.IsExist(ctx, like.VideoID) + if err != nil { + return err + } + if !ok { + return errors.New("video not found") + } + } + + isLiked, err := s.repo.IsLiked(ctx, like.VideoID, like.AccountID) + if err != nil { + return err + } + if isLiked { + return errors.New("user has liked this video") + } + like.CreatedAt = time.Now() - return s.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - if err := tx.Select("id").First(&Video{}, like.VideoID).Error; err != nil { - if errors.Is(err, gorm.ErrRecordNotFound) { - return errors.New("video not found") - } - return err + mysqlEnqueued := false + redisEnqueued := false + if s.likeMQ != nil { + if err := s.likeMQ.Like(ctx, like.AccountID, like.VideoID); err == nil { + mysqlEnqueued = true } - if err := tx.Create(like).Error; err != nil { - if isDupKey(err) { - return errors.New("user has liked this video") - } - return err - } - if err := tx.Model(&Video{}).Where("id = ?", like.VideoID). - UpdateColumn("likes_count", gorm.Expr("likes_count + 1")).Error; err != nil { - return err + } + if s.popularityMQ != nil { + if err := s.popularityMQ.Update(ctx, like.VideoID, 1); err == nil { + redisEnqueued = true } + } + if mysqlEnqueued && redisEnqueued { return nil - }) + } + + // Fallback: direct MySQL write when like MQ publish fails. + if !mysqlEnqueued { + err := s.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Select("id").First(&Video{}, like.VideoID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return errors.New("video not found") + } + return err + } + if err := tx.Create(like).Error; err != nil { + if isDupKey(err) { + return errors.New("user has liked this video") + } + return err + } + if err := tx.Model(&Video{}).Where("id = ?", like.VideoID). + UpdateColumn("likes_count", gorm.Expr("likes_count + 1")).Error; err != nil { + return err + } + return tx.Model(&Video{}).Where("id = ?", like.VideoID). + UpdateColumn("popularity", gorm.Expr("popularity + 1")).Error + }) + if err != nil { + return err + } + } + + // Fallback: direct Redis update when popularity MQ publish fails. + if !redisEnqueued { + UpdatePopularityCache(ctx, s.cache, like.VideoID, 1) + } + return nil } func (s *LikeService) Unlike(ctx context.Context, like *Like) error { - return s.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - del := tx.Where("video_id = ? AND account_id = ?", like.VideoID, like.AccountID).Delete(&Like{}) - if del.Error != nil { - return del.Error - } - if del.RowsAffected == 0 { - return errors.New("user has not liked this video") - } + if like == nil { + return errors.New("like is nil") + } + if like.VideoID == 0 || like.AccountID == 0 { + return errors.New("video_id and account_id are required") + } - return tx.Model(&Video{}).Where("id = ?", like.VideoID). - UpdateColumn("likes_count", gorm.Expr("GREATEST(likes_count - 1, 0)")).Error - }) + if s.VideoRepo != nil { + ok, err := s.VideoRepo.IsExist(ctx, like.VideoID) + if err != nil { + return err + } + if !ok { + return errors.New("video not found") + } + } + + isLiked, err := s.repo.IsLiked(ctx, like.VideoID, like.AccountID) + if err != nil { + return err + } + if !isLiked { + return errors.New("user has not liked this video") + } + + mysqlEnqueued := false + redisEnqueued := false + if s.likeMQ != nil { + if err := s.likeMQ.Unlike(ctx, like.AccountID, like.VideoID); err == nil { + mysqlEnqueued = true + } + } + if s.popularityMQ != nil { + if err := s.popularityMQ.Update(ctx, like.VideoID, -1); err == nil { + redisEnqueued = true + } + } + if mysqlEnqueued && redisEnqueued { + return nil + } + + // Fallback: direct MySQL write when like MQ publish fails. + if !mysqlEnqueued { + err := s.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + del := tx.Where("video_id = ? AND account_id = ?", like.VideoID, like.AccountID).Delete(&Like{}) + if del.Error != nil { + return del.Error + } + if del.RowsAffected == 0 { + return errors.New("user has not liked this video") + } + + if err := tx.Model(&Video{}).Where("id = ?", like.VideoID). + UpdateColumn("likes_count", gorm.Expr("GREATEST(likes_count - 1, 0)")).Error; err != nil { + return err + } + return tx.Model(&Video{}).Where("id = ?", like.VideoID). + UpdateColumn("popularity", gorm.Expr("GREATEST(popularity - 1, 0)")).Error + }) + if err != nil { + return err + } + } + + // Fallback: direct Redis update when popularity MQ publish fails. + if !redisEnqueued { + UpdatePopularityCache(ctx, s.cache, like.VideoID, -1) + } + return nil } func (s *LikeService) IsLiked(ctx context.Context, videoID, accountID uint) (bool, error) { diff --git a/backend/internal/worker/likeworker.go b/backend/internal/worker/likeworker.go new file mode 100644 index 0000000..25a9c6a --- /dev/null +++ b/backend/internal/worker/likeworker.go @@ -0,0 +1,136 @@ +package worker + +import ( + "context" + "encoding/json" + "errors" + "feedsystem_video_go/internal/middleware/rabbitmq" + "feedsystem_video_go/internal/video" + "log" + amqp "github.com/rabbitmq/amqp091-go" + "time" +) + +type LikeWorker struct { + ch *amqp.Channel + likes *video.LikeRepository + videos *video.VideoRepository + queue string +} + +func NewLikeWorker(ch *amqp.Channel, likes *video.LikeRepository, videos *video.VideoRepository, queue string) *LikeWorker { + return &LikeWorker{ch: ch, likes: likes, videos: videos, queue: queue} +} + +func (w *LikeWorker) Run(ctx context.Context) error { + if w == nil || w.ch == nil || w.likes == nil || w.videos == nil { + return errors.New("like 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 *LikeWorker) handleDelivery(ctx context.Context, d amqp.Delivery) { + if err := w.process(ctx, d.Body); err != nil { + log.Printf("like worker: failed to process message: %v", err) + _ = d.Nack(false, true) + return + } + _ = d.Ack(false) +} + +func (w *LikeWorker) process(ctx context.Context, body []byte) error { + var evt rabbitmq.LikeEvent + if err := json.Unmarshal(body, &evt); err != nil { + // 解析事件失败，直接丢弃 + return nil + } + if evt.UserID == 0 || evt.VideoID == 0 { + return nil + } + + switch evt.Action { + case "like": + return w.applyLike(ctx, evt.UserID, evt.VideoID) + case "unlike": + return w.applyUnlike(ctx, evt.UserID, evt.VideoID) + default: + return nil + } +} + +func (w *LikeWorker) applyLike(ctx context.Context, userID, videoID uint) error { + ok, err := w.videos.IsExist(ctx, videoID) + if err != nil { + return err + } + if !ok { + return nil + } + + created, err := w.likes.LikeIgnoreDuplicate(ctx, &video.Like{ + VideoID: videoID, + AccountID: userID, + CreatedAt: time.Now(), + }) + if err != nil { + return err + } + if !created { + return nil + } + + if err := w.videos.ChangeLikesCount(ctx, videoID, 1); err != nil { + return err + } + return w.videos.ChangePopularity(ctx, videoID, 1) +} + +func (w *LikeWorker) applyUnlike(ctx context.Context, userID, videoID uint) error { + ok, err := w.videos.IsExist(ctx, videoID) + if err != nil { + return err + } + if !ok { + return nil + } + + deleted, err := w.likes.DeleteByVideoAndAccount(ctx, videoID, userID) + if err != nil { + return err + } + if !deleted { + return nil + } + + if err := w.videos.ChangeLikesCount(ctx, videoID, -1); err != nil { + return err + } + return w.videos.ChangePopularity(ctx, videoID, -1) +}