diff --git a/internal/constant/constant.go b/internal/constant/constant.go index 74855ed..24bcd93 100644 --- a/internal/constant/constant.go +++ b/internal/constant/constant.go @@ -99,6 +99,8 @@ const ( EventTaskSuccess = "task_success" EventTaskFailed = "task_failed" EventTaskTimeout = "task_timeout" + EventTaskRunning = "task_running" + EventTaskQueued = "task_queued" // 其他事件类型 EventSystemNotice = "system_notice" diff --git a/internal/controllers/system_ws_controller.go b/internal/controllers/system_ws_controller.go new file mode 100644 index 0000000..cf73ccc --- /dev/null +++ b/internal/controllers/system_ws_controller.go @@ -0,0 +1,90 @@ +package controllers + +import ( + "time" + + "github.com/engigu/baihu-panel/internal/constant" + "github.com/engigu/baihu-panel/internal/logger" + "github.com/engigu/baihu-panel/internal/services" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" +) + +type SystemWSController struct { + manager *services.SystemWSManager +} + +func NewSystemWSController() *SystemWSController { + return &SystemWSController{ + manager: services.GetSystemWSManager(), + } +} + +func (sc *SystemWSController) HandleEvents(c *gin.Context) { + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + logger.Errorf("[SystemWS] 升级 WebSocket 失败: %v", err) + return + } + + client := sc.manager.Register(conn) + defer sc.manager.Unregister(client) + + // 启动写循环 + go sc.writeLoop(client) + + // 启动读循环 (主要用于检测连接断开和维持心跳) + sc.readLoop(client) +} + +func (sc *SystemWSController) readLoop(client *services.ClientConnection) { + defer client.Close() + + client.Conn.SetReadLimit(constant.MaxMessageSize) + client.Conn.SetReadDeadline(time.Now().Add(constant.PongWait)) + client.Conn.SetPongHandler(func(string) error { + client.Conn.SetReadDeadline(time.Now().Add(constant.PongWait)) + return nil + }) + + for { + _, _, err := client.Conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + logger.Warnf("[SystemWS] 客户端异常断开: %v", err) + } + break + } + // 暂时不处理来自前端的消息,前端仅作为接收方 + } +} + +func (sc *SystemWSController) writeLoop(client *services.ClientConnection) { + ticker := time.NewTicker(constant.PingPeriod) + defer func() { + ticker.Stop() + client.Close() + }() + + for { + select { + case message, ok := <-client.Send: + client.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if !ok { + // 通道关闭 + client.Conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + if err := client.Conn.WriteMessage(websocket.TextMessage, message); err != nil { + return + } + case <-ticker.C: + client.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if err := client.Conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} diff --git a/internal/models/vo/ws_vo.go b/internal/models/vo/ws_vo.go new file mode 100644 index 0000000..8945799 --- /dev/null +++ b/internal/models/vo/ws_vo.go @@ -0,0 +1,8 @@ +package vo + +// WSMessage 通用 WebSocket 消息结构 +type WSMessage struct { + Type string `json:"type"` // 事件类型: task_status, notice, system_stats + Timestamp int64 `json:"timestamp"` // 毫秒时间戳 + Payload interface{} `json:"payload"` // 负载数据 +} diff --git a/internal/router/api_routes.go b/internal/router/api_routes.go index d531564..18f6b4f 100644 --- a/internal/router/api_routes.go +++ b/internal/router/api_routes.go @@ -61,6 +61,7 @@ func initAuthorizedAPIRoutes(api *gin.RouterGroup, c *Controllers) { registerMiseRoutes(adminOnly, c) registerNotificationRoutes(adminOnly, c) registerAppLogRoutes(adminOnly, c) + registerSystemWSRoutes(adminOnly, c) } } @@ -256,6 +257,10 @@ func registerAppLogRoutes(g *gin.RouterGroup, c *Controllers) { } } +func registerSystemWSRoutes(g *gin.RouterGroup, c *Controllers) { + g.GET("/ws/events", c.SystemWS.HandleEvents) +} + func initAgentAPIRoutes(root *gin.RouterGroup, c *Controllers) { // Agent API(供远程 Agent 调用,不使用 /v1 版本号) agentAPI := root.Group("/api/agent") diff --git a/internal/router/register.go b/internal/router/register.go index 1f5ed81..c380c90 100644 --- a/internal/router/register.go +++ b/internal/router/register.go @@ -23,6 +23,7 @@ func RegisterControllers() *Controllers { scriptService := services.NewScriptService() sendStatsService := services.NewSendStatsService() agentWSManager := services.GetAgentWSManager() + systemWSManager := services.GetSystemWSManager() taskLogService := tasks.NewTaskLogService(sendStatsService) // 创建任务执行服务(需要依赖注入) @@ -40,7 +41,7 @@ func RegisterControllers() *Controllers { executorService.StartCron() // 初始化所有关注系统总线的服务 - setupEventHandlers(appLogService, notifyService, loginLogService) + setupEventHandlers(appLogService, notifyService, loginLogService, systemWSManager) go startAppLogCleanup(appLogService) // 初始化并返回控制器 @@ -61,6 +62,7 @@ func RegisterControllers() *Controllers { Mise: controllers.NewMiseController(services.NewMiseService()), Notification: controllers.NewNotificationController(), AppLog: controllers.NewAppLogController(), + SystemWS: controllers.NewSystemWSController(), } } diff --git a/internal/router/router.go b/internal/router/router.go index 6f1747d..f882665 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -29,6 +29,7 @@ type Controllers struct { Mise *controllers.MiseController Notification *controllers.NotificationController AppLog *controllers.AppLogController + SystemWS *controllers.SystemWSController } func Setup(c *Controllers) *gin.Engine { diff --git a/internal/services/system_ws_service.go b/internal/services/system_ws_service.go new file mode 100644 index 0000000..df901ea --- /dev/null +++ b/internal/services/system_ws_service.go @@ -0,0 +1,126 @@ +package services + +import ( + "encoding/json" + "sync" + "time" + + "github.com/engigu/baihu-panel/internal/constant" + "github.com/engigu/baihu-panel/internal/eventbus" + "github.com/engigu/baihu-panel/internal/logger" + "github.com/engigu/baihu-panel/internal/models/vo" + + "github.com/gorilla/websocket" +) + +// SystemWSManager 前端系统事件 WebSocket 管理器 (单例) +type SystemWSManager struct { + clients map[*ClientConnection]bool + mu sync.RWMutex +} + +// ClientConnection 代表一个前端页面的 WebSocket 连接 +type ClientConnection struct { + Conn *websocket.Conn + Send chan []byte + closed bool + mu sync.Mutex +} + +var systemWSManager *SystemWSManager +var systemWSOnce sync.Once + +// GetSystemWSManager 获取系统 WebSocket 管理器单例 +func GetSystemWSManager() *SystemWSManager { + systemWSOnce.Do(func() { + systemWSManager = &SystemWSManager{ + clients: make(map[*ClientConnection]bool), + } + }) + return systemWSManager +} + +// Register 注册一个新的前端连接 +func (m *SystemWSManager) Register(conn *websocket.Conn) *ClientConnection { + m.mu.Lock() + defer m.mu.Unlock() + + client := &ClientConnection{ + Conn: conn, + Send: make(chan []byte, 256), + } + m.clients[client] = true + return client +} + +// Unregister 注销一个前端连接 +func (m *SystemWSManager) Unregister(client *ClientConnection) { + m.mu.Lock() + defer m.mu.Unlock() + + if _, ok := m.clients[client]; ok { + delete(m.clients, client) + client.Close() + } +} + +// Broadcast 广播消息给所有在线前端 +func (m *SystemWSManager) Broadcast(msgType string, payload interface{}) { + msg := vo.WSMessage{ + Type: msgType, + Timestamp: time.Now().UnixMilli(), + Payload: payload, + } + + data, err := json.Marshal(msg) + if err != nil { + logger.Errorf("[SystemWS] 序列化消息失败: %v", err) + return + } + + m.mu.RLock() + defer m.mu.RUnlock() + + for client := range m.clients { + select { + case client.Send <- data: + default: + // 缓冲区满,可能该客户端连接已死 + go m.Unregister(client) + } + } +} + +// SubscribeEvents 订阅系统事件总线并分发给 WebSocket +func (m *SystemWSManager) SubscribeEvents(bus *eventbus.EventBus) { + // 任务相关事件 + taskEvents := []string{ + constant.EventTaskSuccess, + constant.EventTaskFailed, + constant.EventTaskTimeout, + constant.EventTaskRunning, + constant.EventTaskQueued, + } + + for _, evt := range taskEvents { + bus.Subscribe(evt, func(e eventbus.Event) { + m.Broadcast(e.Type, e.Payload) + }) + } + + // 系统通知事件 + bus.Subscribe(constant.EventSystemNotice, func(e eventbus.Event) { + m.Broadcast("notice", e.Payload) + }) +} + +func (c *ClientConnection) Close() { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return + } + c.closed = true + c.Conn.Close() + close(c.Send) +} diff --git a/internal/services/tasks/executor_service.go b/internal/services/tasks/executor_service.go index e80757f..3333232 100644 --- a/internal/services/tasks/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -116,7 +116,15 @@ type ServerSchedulerHandler struct { } func (h *ServerSchedulerHandler) OnTaskScheduled(req *executor.ExecutionRequest) { - // 任务入队事件,可以在此处更新数据库状态为 "pending" + if req.TaskID != "" { + eventbus.DefaultBus.Publish(eventbus.Event{ + Type: constant.EventTaskQueued, + Payload: map[string]interface{}{ + "task_id": req.TaskID, + "status": constant.TaskStatusQueued, + }, + }) + } } func (h *ServerSchedulerHandler) OnTaskExecuting(req *executor.ExecutionRequest) (io.Writer, io.Writer, error) { @@ -167,6 +175,16 @@ func (h *ServerSchedulerHandler) OnTaskExecuting(req *executor.ExecutionRequest) StartTime: time.Now(), }) + // 发布任务开始运行事件 + eventbus.DefaultBus.Publish(eventbus.Event{ + Type: constant.EventTaskRunning, + Payload: map[string]interface{}{ + "task_id": req.TaskID, + "status": constant.TaskStatusRunning, + "log_id": req.LogID, + }, + }) + if req.Metadata.RetryIndex > 0 { tl.Write([]byte(fmt.Sprintf("\n[System] 此为任务失败后的第 %d 次重试执行...\n\n", req.Metadata.RetryIndex))) } diff --git a/web/public/workers/event-worker.js b/web/public/workers/event-worker.js new file mode 100644 index 0000000..e46a3f2 --- /dev/null +++ b/web/public/workers/event-worker.js @@ -0,0 +1,109 @@ +/** + * Baihu Panel System Event SharedWorker + * + * 这个 Worker 维护全浏览器唯一的一个 WebSocket 连接, + * 并通过 postMessage 将收到的事件同步给所有打开的标签页。 + */ + +let ws = null; +const ports = new Set(); +let reconnectTimer = null; +let lockReconnect = false; + +// WebSocket 配置 (由第一个连接的页面通过消息传过来,或者使用默认值) +let wsUrl = ''; + +/** + * 建立 WebSocket 连接 + */ +function connect() { + if (!wsUrl || ws) return; + + try { + ws = new WebSocket(wsUrl); + + ws.onopen = () => { + console.log('[SharedWorker] WebSocket 已连接'); + lockReconnect = false; + }; + + ws.onmessage = (event) => { + // 广播消息给所有连接的端口 + broadcast(event.data); + }; + + ws.onclose = () => { + console.log('[SharedWorker] WebSocket 已断开'); + ws = null; + reconnect(); + }; + + ws.onerror = (err) => { + console.error('[SharedWorker] WebSocket 错误:', err); + ws = null; + reconnect(); + }; + } catch (e) { + console.error('[SharedWorker] 建立连接失败:', e); + reconnect(); + } +} + +/** + * 重连逻辑 + */ +function reconnect() { + if (lockReconnect) return; + lockReconnect = true; + + if (reconnectTimer) clearTimeout(reconnectTimer); + reconnectTimer = setTimeout(() => { + connect(); + lockReconnect = false; + }, 5000); // 5秒后尝试重连 +} + +/** + * 广播消息给所有活跃标签页 + */ +function broadcast(data) { + const msg = typeof data === 'string' ? JSON.parse(data) : data; + ports.forEach(port => { + try { + port.postMessage(msg); + } catch (e) { + // 如果端口失效,移除它 + ports.delete(port); + } + }); +} + +/** + * 处理 SharedWorker 连接 + */ +self.onconnect = (e) => { + const port = e.ports[0]; + ports.add(port); + + port.onmessage = (event) => { + const { type, data } = event.data; + + switch (type) { + case 'init': + // 初始化配置 + if (data.url) { + wsUrl = data.url; + if (!ws) connect(); + } + break; + case 'ping': + port.postMessage({ type: 'pong' }); + break; + } + }; + + port.start(); + + // 发送一条初始消息确认连接成功 + port.postMessage({ type: 'worker_ready' }); +}; diff --git a/web/src/composables/useEventBus.ts b/web/src/composables/useEventBus.ts new file mode 100644 index 0000000..087f6cd --- /dev/null +++ b/web/src/composables/useEventBus.ts @@ -0,0 +1,35 @@ +import { onMounted, onUnmounted } from 'vue'; +import { eventBus, type WSMessage } from '../utils/event-bus'; + +/** + * Vue Composable: 使用系统事件总线 + * + * @param eventType 可选,指定监听的消息类型 + * @param handler 消息处理回调 + */ +export function useEventBus(eventType: string | string[] | null, handler: (payload: any, type: string) => void) { + let unsubscribe: (() => void) | null = null; + + const handleMessage = (msg: WSMessage) => { + if (!eventType) { + // 监听所有消息 + handler(msg.payload, msg.type); + return; + } + + const types = Array.isArray(eventType) ? eventType : [eventType]; + if (types.includes(msg.type)) { + handler(msg.payload, msg.type); + } + }; + + onMounted(() => { + // 确保驱动已初始化 (单例内部会处理多次调用) + eventBus.init(); + unsubscribe = eventBus.subscribe(handleMessage); + }); + + onUnmounted(() => { + if (unsubscribe) unsubscribe(); + }); +} diff --git a/web/src/utils/event-bus.ts b/web/src/utils/event-bus.ts new file mode 100644 index 0000000..6b0c4bc --- /dev/null +++ b/web/src/utils/event-bus.ts @@ -0,0 +1,129 @@ +/** + * 系统事件总线驱动 (带 SharedWorker 降级逻辑) + */ + +export interface WSMessage { + type: string; + timestamp: number; + payload: any; +} + +export type MessageHandler = (msg: WSMessage) => void; + +class EventBusDriver { + private wsUrl: string; + private handlers: Set = new Set(); + private worker: SharedWorker | null = null; + private socket: WebSocket | null = null; + private reconnectTimer: any = null; + + constructor() { + // 获取基础路径配置 + const baseUrl = (window as any).__BASE_URL__ || ''; + const apiVersion = (window as any).__API_VERSION__ || '/api/v1'; + + // 自动计算 WebSocket 地址 + let host = window.location.host; + let protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + + // 如果 BASE_URL 是绝对路径 (如 http://api.example.com) + if (baseUrl.startsWith('http')) { + const url = new URL(baseUrl); + host = url.host; + protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + } + + // 拼接最终的 WS 地址 + const path = baseUrl.startsWith('http') ? '' : baseUrl; + this.wsUrl = `${protocol}//${host}${path}${apiVersion}/ws/events`; + + // 记录基础路径用于 worker 加载 + this.workerPath = `${path}/workers/event-worker.js`.replace(/\/+/g, '/'); + } + + private workerPath: string; + + private initialized = false; + + /** + * 初始化连接 + */ + init() { + if (this.initialized) return; + + if (window.SharedWorker) { + this.initSharedWorker(); + } else { + this.initStandardWebSocket(); + } + this.initialized = true; + } + + private initSharedWorker() { + try { + this.worker = new SharedWorker(this.workerPath); + this.worker.port.postMessage({ + type: 'init', + data: { url: this.wsUrl } + }); + + this.worker.port.onmessage = (e) => { + this.emit(e.data); + }; + + this.worker.port.start(); + console.log('[EventBus] SharedWorker initialized'); + } catch (e) { + console.error('[EventBus] SharedWorker failed, falling back', e); + this.initStandardWebSocket(); + } + } + + private initStandardWebSocket() { + if (this.socket) return; + + this.socket = new WebSocket(this.wsUrl); + + this.socket.onmessage = (e) => { + try { + const data = JSON.parse(e.data); + this.emit(data); + } catch (err) { + console.error('[EventBus] Parse message error', err); + } + }; + + this.socket.onclose = () => { + this.socket = null; + this.reconnect(); + }; + + this.socket.onerror = () => { + this.socket = null; + this.reconnect(); + }; + + console.log('[EventBus] Standard WebSocket initialized (Fallback)'); + } + + private reconnect() { + if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + this.reconnectTimer = setTimeout(() => this.init(), 5000); + } + + private emit(msg: WSMessage) { + if (!msg || !msg.type) return; + this.handlers.forEach(handler => handler(msg)); + } + + /** + * 订阅消息 + */ + subscribe(handler: MessageHandler) { + this.handlers.add(handler); + return () => this.handlers.delete(handler); + } +} + +// 导出单例 +export const eventBus = new EventBusDriver(); diff --git a/web/src/views/tasks/Tasks.vue b/web/src/views/tasks/Tasks.vue index 81819f8..58ecbd2 100644 --- a/web/src/views/tasks/Tasks.vue +++ b/web/src/views/tasks/Tasks.vue @@ -24,6 +24,7 @@ import { Label } from '@/components/ui/label' import { api, type Agent, type Task, type TaskLog } from '@/api' import { toast } from 'vue-sonner' import { useSiteSettings } from '@/composables/useSiteSettings' +import { useEventBus } from '@/composables/useEventBus' import { useRouter, useRoute } from 'vue-router' import { TASK_TYPE, AGENT_STATUS, TRIGGER_TYPE, TASK_STATUS } from '@/constants' import TextOverflow from '@/components/TextOverflow.vue' @@ -275,7 +276,9 @@ function cleanupLogSocket() { logSocket.onmessage = null logSocket.onerror = null logSocket.onclose = null - logSocket.close() + if (logSocket.readyState === WebSocket.CONNECTING || logSocket.readyState === WebSocket.OPEN) { + logSocket.close() + } logSocket = null } } @@ -472,6 +475,14 @@ onMounted(async () => { loadViewsFromSettings() }) +// 订阅任务状态实时更新 +useEventBus(['task_running', 'task_queued', 'task_success', 'task_failed', 'task_timeout'], (payload) => { + const task = tasks.value.find(t => t.id === payload.task_id) + if (task) { + task.running_status = payload.status + } +}) + // 监听路由参数变化 watch(() => route.query.agent_id, (newVal: any) => { filterAgentId.value = newVal ? String(newVal) : null @@ -624,6 +635,7 @@ watch(() => route.query.agent_id, (newVal: any) => {
+
#{{ total - (currentPage - 1) * pageSize - index }}
@@ -720,6 +732,7 @@ watch(() => route.query.agent_id, (newVal: any) => {
+
#{{ total - (currentPage - 1) * pageSize - index }}
@@ -789,6 +802,7 @@ watch(() => route.query.agent_id, (newVal: any) => {
+
#{{ total - (currentPage - 1) * pageSize - index }}