feat: 点赞相关rabbitMQ

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)
+}

backend/internal/video/like_handler.go

@@ -7,12 +7,11 @@ import (
 )
 
 type LikeHandler struct {
-	service      *LikeService
+	service *LikeService
-	videoService *VideoService
 }
 
-func NewLikeHandler(service *LikeService, videoService *VideoService) *LikeHandler {
+func NewLikeHandler(service *LikeService) *LikeHandler {
-	return &LikeHandler{service: service, videoService: videoService}
+	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"})
 }
 

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{}).

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
+	repo         *LikeRepository
-	VideoRepo *VideoRepository
+	VideoRepo    *VideoRepository
+	cache        *rediscache.Client
+	likeMQ       *rabbitmq.LikeMQ
+	popularityMQ *rabbitmq.PopularityMQ
 }
 
-func NewLikeService(repo *LikeRepository, videoRepo *VideoRepository) *LikeService {
+func NewLikeService(repo *LikeRepository, videoRepo *VideoRepository, cache *rediscache.Client, likeMQ *rabbitmq.LikeMQ, popularityMQ *rabbitmq.PopularityMQ) *LikeService {
-	return &LikeService{repo: repo, VideoRepo: videoRepo}
+	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 {
+	mysqlEnqueued := false
-		if err := tx.Select("id").First(&Video{}, like.VideoID).Error; err != nil {
+	redisEnqueued := false
-			if errors.Is(err, gorm.ErrRecordNotFound) {
+	if s.likeMQ != nil {
-				return errors.New("video not found")
+		if err := s.likeMQ.Like(ctx, like.AccountID, like.VideoID); err == nil {
-			}
+			mysqlEnqueued = true
-			return err
 		}
-		if err := tx.Create(like).Error; err != nil {
+	}
-			if isDupKey(err) {
+	if s.popularityMQ != nil {
-				return errors.New("user has liked this video")
+		if err := s.popularityMQ.Update(ctx, like.VideoID, 1); err == nil {
-			}
+			redisEnqueued = true
-			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 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 {
+	if like == nil {
-		del := tx.Where("video_id = ? AND account_id = ?", like.VideoID, like.AccountID).Delete(&Like{})
+		return errors.New("like is nil")
-		if del.Error != nil {
+	}
-			return del.Error
+	if like.VideoID == 0 || like.AccountID == 0 {
-		}
+		return errors.New("video_id and account_id are required")
-		if del.RowsAffected == 0 {
+	}
-			return errors.New("user has not liked this video")
-		}
 
-		return tx.Model(&Video{}).Where("id = ?", like.VideoID).
+	if s.VideoRepo != nil {
-			UpdateColumn("likes_count", gorm.Expr("GREATEST(likes_count - 1, 0)")).Error
+		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) {

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)
+}