diff --git a/backend/cmd/worker/main.go b/backend/cmd/worker/main.go index f277bc0..de91005 100644 --- a/backend/cmd/worker/main.go +++ b/backend/cmd/worker/main.go @@ -1,295 +1,315 @@ -package main - -import ( - "context" - "feedsystem_video_go/internal/config" - "feedsystem_video_go/internal/db" - rediscache "feedsystem_video_go/internal/middleware/redis" - "feedsystem_video_go/internal/observability" - "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" -) - -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() { - // 加载配置 - configPath := os.Getenv("CONFIG_PATH") - if configPath == "" { - configPath = "configs/config.yaml" - } - log.Printf("Loading config from %s", configPath) - cfg, usedDefault, err := config.LoadLocalDev(configPath) - if err != nil { - log.Fatalf("Failed to load config: %v", err) - } - if usedDefault { - log.Printf("Config File %s not found, using default local config", configPath) - } else { - log.Printf("Config loaded from file: %s", configPath) - } - // 连接数据库 - sqlDB, err := db.NewDB(cfg.Database) - if err != nil { - 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) - 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 := 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) - 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() - - pprofServer, err := observability.NewPprofServer( - "Worker", - cfg.ObservabilityConfig.Pprof.Enabled, - cfg.ObservabilityConfig.Pprof.WorkerAddr, - ) - if err != nil { - log.Printf("Failed to start worker pprof server: %v", err) - } - if pprofServer != nil { - defer pprofServer.Close() - } - - errCh := make(chan error, 4) - 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("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 -} - -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, - ) -} +package main + +import ( + "context" + "feedsystem_video_go/internal/config" + "feedsystem_video_go/internal/db" + rediscache "feedsystem_video_go/internal/middleware/redis" + "feedsystem_video_go/internal/observability" + "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" + "gorm.io/gorm" +) + +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 connectWithRetry(name string, maxRetries int, fn func() error) { + for i := 0; i < maxRetries; i++ { + if err := fn(); err == nil { + return + } + wait := time.Duration(1< 30*time.Second { + wait = 30 * time.Second + } + log.Printf("%s 不可用，%v 后重试 (%d/%d)...", name, wait, i+1, maxRetries) + time.Sleep(wait) + } + log.Fatalf("%s: 超过最大重试次数", name) +} + +func main() { + // 加载配置 + configPath := os.Getenv("CONFIG_PATH") + if configPath == "" { + configPath = "configs/config.yaml" + } + log.Printf("Loading config from %s", configPath) + cfg, usedDefault, err := config.LoadLocalDev(configPath) + if err != nil { + log.Fatalf("Failed to load config: %v", err) + } + if usedDefault { + log.Printf("Config File %s not found, using default local config", configPath) + } else { + log.Printf("Config loaded from file: %s", configPath) + } + // 连接数据库（带重试） + var sqlDB *gorm.DB + connectWithRetry("MySQL", 10, func() error { + var err error + sqlDB, err = db.NewDB(cfg.Database) + return 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) + "/" + var conn *amqp.Connection + connectWithRetry("RabbitMQ", 10, func() error { + var err error + conn, err = amqp.Dial(url) + return 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 := 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) + 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() + + pprofServer, err := observability.NewPprofServer( + "Worker", + cfg.ObservabilityConfig.Pprof.Enabled, + cfg.ObservabilityConfig.Pprof.WorkerAddr, + ) + if err != nil { + log.Printf("Failed to start worker pprof server: %v", err) + } + if pprofServer != nil { + defer pprofServer.Close() + } + + errCh := make(chan error, 4) + 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("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 +} + +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, + ) +} diff --git a/docker-compose.yml b/docker-compose.yml index 3145017..4d1324e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -71,6 +71,11 @@ services: condition: service_healthy rabbitmq: condition: service_healthy + healthcheck: + test: ["CMD-SHELL", "wget -qO- --post-data='{}' --header='Content-Type: application/json' http://localhost:8080/account/findByID || exit 1"] + interval: 10s + timeout: 5s + retries: 3 worker: build: @@ -87,6 +92,11 @@ services: condition: service_healthy rabbitmq: condition: service_healthy + healthcheck: + test: ["CMD-SHELL", "pgrep worker || exit 1"] + interval: 15s + timeout: 5s + retries: 3 frontend: build: @@ -97,6 +107,11 @@ services: - "5173:80" depends_on: - backend + healthcheck: + test: ["CMD-SHELL", "wget -qO- http://localhost:80/ || exit 1"] + interval: 10s + timeout: 5s + retries: 3 volumes: mysql_data: diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 1531b58..7f4bbd9 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -1,99 +1,104 @@ -import { useAuthStore } from '../stores/auth' - -export class ApiError extends Error { - status: number - payload?: unknown - - constructor(message: string, status: number, payload?: unknown) { - super(message) - this.name = 'ApiError' - this.status = status - this.payload = payload - } -} - -type ApiErrorBody = { error?: string } - -const API_BASE = (import.meta.env.VITE_API_BASE as string | undefined) ?? '/api' - -export async function postJson(path: string, body: unknown, options?: { authRequired?: boolean }): Promise { - const auth = useAuthStore() - const token = auth.token - - if (options?.authRequired && !token) { - throw new ApiError('需要先登录（缺少 token）', 401) - } - - const headers: Record = { 'Content-Type': 'application/json' } - if (token) headers.Authorization = `Bearer ${token}` - - const res = await fetch(`${API_BASE}${path}`, { - method: 'POST', - headers, - body: JSON.stringify(body ?? {}), - }) - - const text = await res.text() - let data: unknown = null - if (text) { - try { - data = JSON.parse(text) - } catch { - data = text - } - } - - if (!res.ok) { - if (res.status === 401) { - auth.clearToken() - } - const msg = - data && typeof data === 'object' && (data as ApiErrorBody).error - ? String((data as ApiErrorBody).error) - : `请求失败 (${res.status})` - throw new ApiError(msg, res.status, data) - } - - return data as T -} - -export async function postForm(path: string, body: FormData, options?: { authRequired?: boolean }): Promise { - const auth = useAuthStore() - const token = auth.token - - if (options?.authRequired && !token) { - throw new ApiError('需要先登录（缺少 token）', 401) - } - - const headers: Record = {} - if (token) headers.Authorization = `Bearer ${token}` - - const res = await fetch(`${API_BASE}${path}`, { - method: 'POST', - headers, - body, - }) - - const text = await res.text() - let data: unknown = null - if (text) { - try { - data = JSON.parse(text) - } catch { - data = text - } - } - - if (!res.ok) { - if (res.status === 401) { - auth.clearToken() - } - const msg = - data && typeof data === 'object' && (data as ApiErrorBody).error - ? String((data as ApiErrorBody).error) - : `请求失败 (${res.status})` - throw new ApiError(msg, res.status, data) - } - - return data as T -} +import { useAuthStore } from '../stores/auth' +import { reportError } from '../utils/error-reporter' + +export class ApiError extends Error { + status: number + payload?: unknown + + constructor(message: string, status: number, payload?: unknown) { + super(message) + this.name = 'ApiError' + this.status = status + this.payload = payload + } +} + +type ApiErrorBody = { error?: string } + +const API_BASE = (import.meta.env.VITE_API_BASE as string | undefined) ?? '/api' + +export async function postJson(path: string, body: unknown, options?: { authRequired?: boolean }): Promise { + const auth = useAuthStore() + const token = auth.token + + if (options?.authRequired && !token) { + throw new ApiError('需要先登录（缺少 token）', 401) + } + + const headers: Record = { 'Content-Type': 'application/json' } + if (token) headers.Authorization = `Bearer ${token}` + + const res = await fetch(`${API_BASE}${path}`, { + method: 'POST', + headers, + body: JSON.stringify(body ?? {}), + }) + + const text = await res.text() + let data: unknown = null + if (text) { + try { + data = JSON.parse(text) + } catch { + data = text + } + } + + if (!res.ok) { + if (res.status === 401) { + auth.clearToken() + } + const msg = + data && typeof data === 'object' && (data as ApiErrorBody).error + ? String((data as ApiErrorBody).error) + : `请求失败 (${res.status})` + const apiErr = new ApiError(msg, res.status, data) + reportError(apiErr, { path, status: res.status }) + throw apiErr + } + + return data as T +} + +export async function postForm(path: string, body: FormData, options?: { authRequired?: boolean }): Promise { + const auth = useAuthStore() + const token = auth.token + + if (options?.authRequired && !token) { + throw new ApiError('需要先登录（缺少 token）', 401) + } + + const headers: Record = {} + if (token) headers.Authorization = `Bearer ${token}` + + const res = await fetch(`${API_BASE}${path}`, { + method: 'POST', + headers, + body, + }) + + const text = await res.text() + let data: unknown = null + if (text) { + try { + data = JSON.parse(text) + } catch { + data = text + } + } + + if (!res.ok) { + if (res.status === 401) { + auth.clearToken() + } + const msg = + data && typeof data === 'object' && (data as ApiErrorBody).error + ? String((data as ApiErrorBody).error) + : `请求失败 (${res.status})` + const apiErr = new ApiError(msg, res.status, data) + reportError(apiErr, { path, status: res.status }) + throw apiErr + } + + return data as T +} diff --git a/frontend/src/main.ts b/frontend/src/main.ts index 6a92eff..d959065 100644 --- a/frontend/src/main.ts +++ b/frontend/src/main.ts @@ -1,10 +1,16 @@ -import { createApp } from 'vue' -import { createPinia } from 'pinia' -import './style.css' -import App from './App.vue' -import router from './router' - -const app = createApp(App) -app.use(createPinia()) -app.use(router) -app.mount('#app') +import { createApp } from 'vue' +import { createPinia } from 'pinia' +import './style.css' +import App from './App.vue' +import router from './router' +import { reportError } from './utils/error-reporter' + +const app = createApp(App) +app.use(createPinia()) +app.use(router) + +app.config.errorHandler = (err, _instance, info) => { + reportError(err instanceof Error ? err : new Error(String(err)), { info }) +} + +app.mount('#app') diff --git a/frontend/src/utils/error-reporter.ts b/frontend/src/utils/error-reporter.ts new file mode 100644 index 0000000..2ebc1e2 --- /dev/null +++ b/frontend/src/utils/error-reporter.ts @@ -0,0 +1,18 @@ +export function reportError(error: Error, context?: Record) { + if (import.meta.env.DEV) { + console.error('[ErrorReporter]', error.message, context) + return + } + fetch('/api/error-report', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + message: error.message, + stack: error.stack, + context, + timestamp: new Date().toISOString(), + }), + }).catch(() => { + /* 静默失败，避免错误上报自身导致循环 */ + }) +}