feat: 添加Social消费者

backend/cmd/worker/main.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+	"context"
+	"feedsystem_video_go/internal/config"
+	"feedsystem_video_go/internal/db"
+	"feedsystem_video_go/internal/social"
+	"feedsystem_video_go/internal/worker"
+	"log"
+	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const (
+	socialExchange   = "social.events"
+	socialQueue      = "social.events"
+	socialBindingKey = "social.*"
+)
+
+func main() {
+	// 加载配置
+	log.Printf("Loading config from configs/config.yaml")
+	cfg, err := config.Load("configs/config.yaml")
+	if err != nil {
+		log.Fatalf("Failed to load config: %v", err)
+	}
+	// 连接数据库
+	sqlDB, err := db.NewDB(cfg.Database)
+	if err != nil {
+		log.Fatalf("Failed to connect database: %v", err)
+	}
+	defer db.CloseDB(sqlDB)
+	// 连接 RabbitMQ
+	url := "amqp://" + cfg.RabbitMQ.Username + ":" + cfg.RabbitMQ.Password + "@" + cfg.RabbitMQ.Host + ":" + strconv.Itoa(cfg.RabbitMQ.Port) + "/"
+	conn, err := amqp.Dial(url)
+	if err != nil {
+		log.Fatalf("Failed to connect rabbitmq: %v", err)
+	}
+	defer conn.Close()
+	// 创建 RabbitMQ 通道
+	ch, err := conn.Channel()
+	if err != nil {
+		log.Fatalf("Failed to open rabbitmq channel: %v", err)
+	}
+	defer ch.Close()
+	// 声明 Social 交换机和队列
+	if err := declareSocialTopology(ch); err != nil {
+		log.Fatalf("Failed to declare social 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)
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	log.Printf("Social worker started, consuming queue=%s", socialQueue)
+	if err := worker.Run(ctx); err != nil && err != context.Canceled {
+		log.Fatalf("Worker stopped: %v", err)
+	}
+	log.Printf("Social worker stopped")
+}
+
+func declareSocialTopology(ch *amqp.Channel) error {
+	if err := ch.ExchangeDeclare(
+		socialExchange,
+		"topic",
+		true,
+		false,
+		false,
+		false,
+		nil,
+	); err != nil {
+		return err
+	}
+
+	q, err := ch.QueueDeclare(
+		socialQueue,
+		true,
+		false,
+		false,
+		false,
+		nil,
+	)
+	if err != nil {
+		return err
+	}
+
+	if err := ch.QueueBind(
+		q.Name,
+		socialBindingKey,
+		socialExchange,
+		false,
+		nil,
+	); err != nil {
+		return err
+	}
+	return nil
+}

backend/internal/worker/socialworker.go

@@ -0,0 +1,101 @@
+package worker
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"feedsystem_video_go/internal/middleware/rabbitmq"
+	"feedsystem_video_go/internal/social"
+	"log"
+
+	"github.com/go-sql-driver/mysql"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type SocialWorker struct {
+	ch    *amqp.Channel
+	repo  *social.SocialRepository
+	queue string
+}
+
+func NewSocialWorker(ch *amqp.Channel, repo *social.SocialRepository, queue string) *SocialWorker {
+	return &SocialWorker{ch: ch, repo: repo, queue: queue}
+}
+
+func (w *SocialWorker) Run(ctx context.Context) error {
+	if w == nil || w.ch == nil || w.repo == nil {
+		return errors.New("social 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 *SocialWorker) handleDelivery(ctx context.Context, d amqp.Delivery) {
+	if err := w.process(ctx, d.Body); err != nil {
+		log.Printf("social worker: failed to process message: %v", err)
+		// Retry on DB/transient errors.
+		_ = d.Nack(false, true)
+		return
+	}
+	_ = d.Ack(false)
+}
+
+func (w *SocialWorker) process(ctx context.Context, body []byte) error {
+	var evt rabbitmq.SocialEvent
+	if err := json.Unmarshal(body, &evt); err != nil {
+		// Poison message: drop it to avoid infinite retry loop.
+		return nil
+	}
+	if evt.FollowerID == 0 || evt.VloggerID == 0 {
+		return nil
+	}
+
+	switch evt.Action {
+	case "follow":
+		err := w.repo.Follow(ctx, &social.Social{
+			FollowerID: evt.FollowerID,
+			VloggerID:  evt.VloggerID,
+		})
+		if err == nil {
+			return nil
+		}
+		var mysqlErr *mysql.MySQLError
+		if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
+			return nil
+		}
+		return err
+	case "unfollow":
+		return w.repo.Unfollow(ctx, &social.Social{
+			FollowerID: evt.FollowerID,
+			VloggerID:  evt.VloggerID,
+		})
+	default:
+		return nil
+	}
+}