diff --git a/backend/cmd/worker/main.go b/backend/cmd/worker/main.go new file mode 100644 index 0000000..c62d7a9 --- /dev/null +++ b/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 +} diff --git a/backend/internal/worker/socialworker.go b/backend/internal/worker/socialworker.go new file mode 100644 index 0000000..00065fd --- /dev/null +++ b/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 + } +}