feat: 评论相关rabbitMQ
This commit is contained in:
Leon
71
backend/internal/middleware/rabbitmq/commentMQ.go
Normal file
71
backend/internal/middleware/rabbitmq/commentMQ.go
Normal file
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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"})
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
122
backend/internal/worker/commentworker.go
Normal file
122
backend/internal/worker/commentworker.go
Normal file
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user