feat: 启动新添加的消息队列

backend/cmd/worker/main.go

@@ -4,13 +4,16 @@ import (
 	"context"
 	"feedsystem_video_go/internal/config"
 	"feedsystem_video_go/internal/db"
+	rediscache "feedsystem_video_go/internal/middleware/redis"
 	"feedsystem_video_go/internal/social"
+	"feedsystem_video_go/internal/video"
 	"feedsystem_video_go/internal/worker"
 	"log"
 	"os"
 	"os/signal"
 	"strconv"
 	"syscall"
+	"time"
 
 	amqp "github.com/rabbitmq/amqp091-go"
 )
@@ -19,6 +22,18 @@ const (
 	socialExchange   = "social.events"
 	socialQueue      = "social.events"
 	socialBindingKey = "social.*"
+
+	likeExchange   = "like.events"
+	likeQueue      = "like.events"
+	likeBindingKey = "like.*"
+
+	commentExchange   = "comment.events"
+	commentQueue      = "comment.events"
+	commentBindingKey = "comment.*"
+
+	popularityExchange   = "video.popularity.events"
+	popularityQueue      = "video.popularity.events"
+	popularityBindingKey = "video.popularity.*"
 )
 
 func main() {
@@ -34,6 +49,24 @@ func main() {
 		log.Fatalf("Failed to connect database: %v", err)
 	}
 	defer db.CloseDB(sqlDB)
+
+	// 连接 Redis（用于流行度更新）
+	cache, err := rediscache.NewFromEnv(&cfg.Redis)
+	if err != nil {
+		log.Printf("Redis config error (popularity worker disabled): %v", err)
+		cache = nil
+	} else {
+		pingCtx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+		defer cancel()
+		if err := cache.Ping(pingCtx); err != nil {
+			log.Printf("Redis not available (popularity worker disabled): %v", err)
+			_ = cache.Close()
+			cache = nil
+		} else {
+			defer cache.Close()
+			log.Printf("Redis connected (popularity worker enabled)")
+		}
+	}
 	// 连接 RabbitMQ
 	url := "amqp://" + cfg.RabbitMQ.Username + ":" + cfg.RabbitMQ.Password + "@" + cfg.RabbitMQ.Host + ":" + strconv.Itoa(cfg.RabbitMQ.Port) + "/"
 	conn, err := amqp.Dial(url)
@@ -51,21 +84,53 @@ func main() {
 	if err := declareSocialTopology(ch); err != nil {
 		log.Fatalf("Failed to declare social topology: %v", err)
 	}
+	if err := declareLikeTopology(ch); err != nil {
+		log.Fatalf("Failed to declare like topology: %v", err)
+	}
+	if err := declareCommentTopology(ch); err != nil {
+		log.Fatalf("Failed to declare comment topology: %v", err)
+	}
+	if cache != nil {
+		if err := declarePopularityTopology(ch); err != nil {
+			log.Fatalf("Failed to declare popularity topology: %v", err)
+		}
+	}
 	if err := ch.Qos(50, 0, false); err != nil {
 		log.Fatalf("Failed to set qos: %v", err)
 	}
 
 	repo := social.NewSocialRepository(sqlDB)
-	worker := worker.NewSocialWorker(ch, repo, socialQueue)
+	socialWorker := worker.NewSocialWorker(ch, repo, socialQueue)
+	videoRepo := video.NewVideoRepository(sqlDB)
+	likeRepo := video.NewLikeRepository(sqlDB)
+	commentRepo := video.NewCommentRepository(sqlDB)
+	likeWorker := worker.NewLikeWorker(ch, likeRepo, videoRepo, likeQueue)
+	commentWorker := worker.NewCommentWorker(ch, commentRepo, videoRepo, commentQueue)
+	var popularityWorker *worker.PopularityWorker
+	if cache != nil {
+		popularityWorker = worker.NewPopularityWorker(ch, cache, popularityQueue)
+	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 
-	log.Printf("Social worker started, consuming queue=%s", socialQueue)
+	errCh := make(chan error, 4)
-	if err := worker.Run(ctx); err != nil && err != context.Canceled {
+	log.Printf("Worker started, consuming queue=%s", socialQueue)
+	go func() { errCh <- socialWorker.Run(ctx) }()
+	log.Printf("Worker started, consuming queue=%s", likeQueue)
+	go func() { errCh <- likeWorker.Run(ctx) }()
+	log.Printf("Worker started, consuming queue=%s", commentQueue)
+	go func() { errCh <- commentWorker.Run(ctx) }()
+	if popularityWorker != nil {
+		log.Printf("Worker started, consuming queue=%s", popularityQueue)
+		go func() { errCh <- popularityWorker.Run(ctx) }()
+	}
+
+	err = <-errCh
+	if err != nil && err != context.Canceled {
 		log.Fatalf("Worker stopped: %v", err)
 	}
-	log.Printf("Social worker stopped")
+	log.Printf("Worker stopped")
 }
 
 func declareSocialTopology(ch *amqp.Channel) error {
@@ -104,3 +169,105 @@ func declareSocialTopology(ch *amqp.Channel) error {
 	}
 	return nil
 }
+
+func declarePopularityTopology(ch *amqp.Channel) error {
+	if err := ch.ExchangeDeclare(
+		popularityExchange,
+		"topic",
+		true,
+		false,
+		false,
+		false,
+		nil,
+	); err != nil {
+		return err
+	}
+
+	q, err := ch.QueueDeclare(
+		popularityQueue,
+		true,
+		false,
+		false,
+		false,
+		nil,
+	)
+	if err != nil {
+		return err
+	}
+
+	return ch.QueueBind(
+		q.Name,
+		popularityBindingKey,
+		popularityExchange,
+		false,
+		nil,
+	)
+}
+
+func declareLikeTopology(ch *amqp.Channel) error {
+	if err := ch.ExchangeDeclare(
+		likeExchange,
+		"topic",
+		true,
+		false,
+		false,
+		false,
+		nil,
+	); err != nil {
+		return err
+	}
+
+	q, err := ch.QueueDeclare(
+		likeQueue,
+		true,
+		false,
+		false,
+		false,
+		nil,
+	)
+	if err != nil {
+		return err
+	}
+
+	return ch.QueueBind(
+		q.Name,
+		likeBindingKey,
+		likeExchange,
+		false,
+		nil,
+	)
+}
+
+func declareCommentTopology(ch *amqp.Channel) error {
+	if err := ch.ExchangeDeclare(
+		commentExchange,
+		"topic",
+		true,
+		false,
+		false,
+		false,
+		nil,
+	); err != nil {
+		return err
+	}
+
+	q, err := ch.QueueDeclare(
+		commentQueue,
+		true,
+		false,
+		false,
+		false,
+		nil,
+	)
+	if err != nil {
+		return err
+	}
+
+	return ch.QueueBind(
+		q.Name,
+		commentBindingKey,
+		commentExchange,
+		false,
+		nil,
+	)
+}