refactor: migrate real-time log viewing from WebSocket to SSE with JSON wrapping

2026-06-29 22:00:09 +08:00
parent 06980beeb6
commit e5fc716b8f
7 changed files with 147 additions and 171 deletions
@@ -0,0 +1,99 @@
+package controllers
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/engigu/baihu-panel/internal/database"
+	"github.com/engigu/baihu-panel/internal/models"
+	"github.com/engigu/baihu-panel/internal/services/tasks"
+	"github.com/engigu/baihu-panel/internal/utils"
+
+	"github.com/gin-gonic/gin"
+)
+
+type LogSSEController struct{}
+
+func NewLogSSEController() *LogSSEController {
+	return &LogSSEController{}
+}
+
+func (lc *LogSSEController) StreamLog(c *gin.Context) {
+	logIDStr := c.Query("log_id")
+	if logIDStr == "" {
+		c.JSON(400, gin.H{"error": "log_id is required"})
+		return
+	}
+
+	logID := logIDStr
+
+	c.Header("Content-Type", "text/event-stream")
+	c.Header("Cache-Control", "no-cache")
+	c.Header("Connection", "keep-alive")
+	c.Header("Transfer-Encoding", "chunked")
+	// c.Header("Access-Control-Allow-Origin", "*")
+
+	// 1. 检查数据库中是否已结束
+	var taskLog models.TaskLog
+	res := database.DB.Where("id = ?", logID).Limit(1).Find(&taskLog)
+	if res.Error == nil && res.RowsAffected > 0 {
+		if taskLog.Status != "running" {
+			// 已结束，读取库内日志
+			content, err := utils.DecompressFromBase64(string(taskLog.Output))
+			if err != nil {
+				c.SSEvent("message", gin.H{"text": "解压日志失败: " + err.Error()})
+				c.Writer.Flush()
+				return
+			}
+			c.SSEvent("message", gin.H{"text": content})
+			c.Writer.Flush()
+			return
+		}
+	}
+
+	// 2. 未结束或未找到记录，尝试从 TinyLogManager 获取
+	tl := tasks.GetActiveLog(logID)
+	if tl == nil {
+		c.SSEvent("message", gin.H{"text": "未找到正在运行的任务日志"})
+		c.Writer.Flush()
+		return
+	}
+
+	// 发送系统提示
+	c.SSEvent("message", gin.H{"text": fmt.Sprintf("[System] 连接成功，正在监听日志... (LogID: %s)\n", logID)})
+	c.Writer.Flush()
+
+	// 发送最后 100 行
+	lastLines, err := tl.ReadLastLines(100)
+	if err == nil && len(lastLines) > 0 {
+		c.SSEvent("message", gin.H{"text": string(lastLines)})
+		c.Writer.Flush()
+	}
+
+	// 订阅实时更新
+	sub := tl.Subscribe()
+	defer tl.Unsubscribe(sub)
+
+	// 推送更新
+	c.Stream(func(w io.Writer) bool {
+		select {
+		case data, ok := <-sub:
+			if !ok {
+				// 任务结束，尝试刷新最后一次库内完整内容
+				var finalLog models.TaskLog
+				res := database.DB.Where("id = ?", logID).Limit(1).Find(&finalLog)
+				if res.Error == nil && res.RowsAffected > 0 {
+					content, _ := utils.DecompressFromBase64(string(finalLog.Output))
+					if content != "" {
+						c.SSEvent("message", gin.H{"text": "\n--- 任务已结束 ---\n"})
+					}
+				}
+				return false
+			}
+			c.SSEvent("message", gin.H{"text": string(data)})
+			return true
+		case <-c.Request.Context().Done():
+			return false
+		}
+	})
+}
@@ -1,122 +0,0 @@
-package controllers
-
-import (
-	"fmt"
-
-	"github.com/engigu/baihu-panel/internal/constant"
-	"github.com/engigu/baihu-panel/internal/database"
-	"github.com/engigu/baihu-panel/internal/models"
-	"github.com/engigu/baihu-panel/internal/services/tasks"
-	"github.com/engigu/baihu-panel/internal/utils"
-
-	"github.com/gin-gonic/gin"
-	"github.com/gorilla/websocket"
-	"time"
-)
-
-type LogWSController struct{}
-
-func NewLogWSController() *LogWSController {
-	return &LogWSController{}
-}
-
-func (lc *LogWSController) StreamLog(c *gin.Context) {
-	logIDStr := c.Query("log_id")
-	if logIDStr == "" {
-		return
-	}
-
-	logID := logIDStr
-
-	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
-	if err != nil {
-		return
-	}
-	defer conn.Close()
-
-	// DoS 保护与心跳设置
-	conn.SetReadLimit(constant.MaxMessageSize) // 使用与终端一致的限制
-	conn.SetReadDeadline(time.Now().Add(constant.PongWait))
-	conn.SetPongHandler(func(string) error {
-		conn.SetReadDeadline(time.Now().Add(constant.PongWait))
-		return nil
-	})
-
-	// 启动一个读取循环，用于处理 pong 和检测断开
-	go func() {
-		for {
-			if _, _, err := conn.ReadMessage(); err != nil {
-				break
-			}
-		}
-	}()
-
-	// 1. 检查数据库中是否已结束
-	var taskLog models.TaskLog
-	res := database.DB.Where("id = ?", logID).Limit(1).Find(&taskLog)
-	if res.Error == nil && res.RowsAffected > 0 {
-		if taskLog.Status != "running" {
-			// 已结束，读取库内日志
-			content, err := utils.DecompressFromBase64(string(taskLog.Output))
-			if err != nil {
-				conn.WriteMessage(websocket.TextMessage, []byte("解压日志失败: "+err.Error()))
-				return
-			}
-			conn.WriteMessage(websocket.TextMessage, []byte(content))
-			return
-		}
-	}
-
-	// 2. 未结束或未找到记录，尝试从 TinyLogManager 获取
-	tl := tasks.GetActiveLog(logID)
-	if tl == nil {
-		conn.WriteMessage(websocket.TextMessage, []byte("未找到正在运行的任务日志"))
-		return
-	}
-
-	// 发送系统提示
-	conn.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf("[System] 连接成功，正在监听日志... (LogID: %s)\n", logID)))
-
-	// 发送最后 100 行
-	lastLines, err := tl.ReadLastLines(100)
-	if err == nil && len(lastLines) > 0 {
-		conn.WriteMessage(websocket.TextMessage, lastLines)
-	}
-
-	// 订阅实时更新
-	sub := tl.Subscribe()
-	defer tl.Unsubscribe(sub)
-
-	ticker := time.NewTicker(constant.PingPeriod)
-	defer ticker.Stop()
-
-	// 推送更新
-	for {
-		select {
-		case <-ticker.C:
-			conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
-			if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
-				return
-			}
-		case data, ok := <-sub:
-			if !ok {
-				// 任务结束，尝试刷新最后一次库内完整内容
-				var finalLog models.TaskLog
-				res := database.DB.Where("id = ?", logID).Limit(1).Find(&finalLog)
-				if res.Error == nil && res.RowsAffected > 0 {
-					content, _ := utils.DecompressFromBase64(string(finalLog.Output))
-					if content != "" {
-						conn.WriteMessage(websocket.TextMessage, []byte("\n--- 任务已结束 ---\n"))
-						// 这里可以选择性再推一次完整版，或直接退出
-					}
-				}
-				return
-			}
-			if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
-				return
-			}
-		case <-c.Request.Context().Done():
-			return
-		}
-	}
-}
@@ -161,7 +161,7 @@ func registerLogRoutes(g *gin.RouterGroup, c *Controllers) {
 	{
 		logs.GET("", c.Log.GetLogs)
 		logs.POST("/clear", c.Log.ClearLogs)
-		logs.GET("/ws", c.LogWS.StreamLog)
+		logs.GET("/sse", c.LogSSE.StreamLog)
 		logs.GET("/:id", c.Log.GetLogDetail)
 		logs.DELETE("/:id", c.Log.DeleteLog)
 	}
@@ -58,7 +58,7 @@ func RegisterControllers() *Controllers {
 		File:         controllers.NewFileController(constant.ScriptsWorkDir),
 		Dashboard:    controllers.NewDashboardController(executorService),
 		Log:          controllers.NewLogController(),
-		LogWS:        controllers.NewLogWSController(),
+		LogSSE:       controllers.NewLogSSEController(),
 		Terminal:     controllers.NewTerminalController(envService),
 		Settings:     controllers.NewSettingsController(userService, loginLogService, executorService),
 		Dependency:   controllers.NewDependencyController(),
@@ -21,7 +21,7 @@ type Controllers struct {
 	File         *controllers.FileController
 	Dashboard    *controllers.DashboardController
 	Log          *controllers.LogController
-	LogWS        *controllers.LogWSController
+	LogSSE       *controllers.LogSSEController
 	Terminal     *controllers.TerminalController
 	Settings     *controllers.SettingsController
 	Dependency   *controllers.DependencyController
@@ -80,7 +80,7 @@ const deleteLogId = ref<string | null>(null)

 const wsContent = ref('')
 const isWsLoading = ref(false)
-let logSocket: WebSocket | null = null
+let logSource: EventSource | null = null


 import { decompressFromBase64 } from '@/utils/decompress'
@@ -136,12 +136,9 @@ function handlePageChange(page: number) {
 }

 async function selectLog(log: TaskLog) {
-  if (logSocket) {
-    logSocket.onopen = null
-    logSocket.onmessage = null
-    logSocket.onerror = null
-    logSocket.onclose = null
-    logSocket.close()
+  if (logSource) {
+    logSource.close()
+    logSource = null
  }

  // 清理旧定时器
@@ -198,23 +195,28 @@ async function selectLog(log: TaskLog) {
    }
    return
  }
-
-  const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
+  
+  const protocol = window.location.protocol
  const host = window.location.host
  const baseUrl = (window as any).__BASE_URL__ || ''
  const apiVersion = (window as any).__API_VERSION__ || '/api/v1'
-  const wsUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/ws?log_id=${log.id}`
+  const sseUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/sse?log_id=${log.id}`

-  logSocket = new WebSocket(wsUrl)
+  logSource = new EventSource(sseUrl)

-  logSocket.onopen = () => {
+  logSource.onopen = () => {
    isWsLoading.value = false
-    console.log('[LogWS] Connection opened')
+    console.log('[LogSSE] Connection opened')
  }

-  logSocket.onmessage = (event) => {
+  logSource.onmessage = (event) => {
    isWsLoading.value = false
-    wsContent.value += event.data
+    try {
+      const data = JSON.parse(event.data)
+      wsContent.value += data.text || ''
+    } catch {
+      wsContent.value += event.data
+    }
    // 自动滚动到底部
    nextTick(() => {
      const pre = document.querySelector('.log-pre')
@@ -222,15 +224,13 @@ async function selectLog(log: TaskLog) {
    })
  }

-  logSocket.onerror = (e) => {
+  logSource.onerror = (e) => {
    isWsLoading.value = false
-    console.error('[LogWS] Connection error', e)
-    toast.error('日志连接异常')
-  }
-
-  logSocket.onclose = (e) => {
-    isWsLoading.value = false
-    console.log('[LogWS] Connection closed', e.code, e.reason)
+    console.error('[LogSSE] Connection error/closed', e)
+    if (logSource) {
+      logSource.close()
+      logSource = null
+    }
  }
 }

@@ -239,13 +239,9 @@ function closeDetail() {
    clearInterval(durationTimer)
    durationTimer = null
  }
-  if (logSocket) {
-    logSocket.onopen = null
-    logSocket.onmessage = null
-    logSocket.onerror = null
-    logSocket.onclose = null
-    logSocket.close()
-    logSocket = null
+  if (logSource) {
+    logSource.close()
+    logSource = null
  }
  selectedLog.value = null
  wsContent.value = ''
@@ -295,19 +295,13 @@ const selectedLog = ref<TaskLog | null>(null)
 const logContent = ref('')
 const logEmptyTitle = ref<string | undefined>(undefined)
 const logEmptyDesc = ref<string | undefined>(undefined)
-let logSocket: WebSocket | null = null
+let logSource: EventSource | null = null
 let durationTimer: ReturnType<typeof setInterval> | null = null

 function cleanupLogSocket() {
-  if (logSocket) {
-    logSocket.onopen = null
-    logSocket.onmessage = null
-    logSocket.onerror = null
-    logSocket.onclose = null
-    if (logSocket.readyState === WebSocket.CONNECTING || logSocket.readyState === WebSocket.OPEN) {
-      logSocket.close()
-    }
-    logSocket = null
+  if (logSource) {
+    logSource.close()
+    logSource = null
  }
 }

@@ -360,17 +354,26 @@ async function viewLogs(taskId: string) {
        return
      }

-      // Connect WebSocket to load log content for running tasks
+      // Connect SSE to load log content for running tasks
      cleanupLogSocket()
-      const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
+      const protocol = window.location.protocol
      const host = window.location.host
      const baseUrl = (window as any).__BASE_URL__ || ''
      const apiVersion = (window as any).__API_VERSION__ || '/api/v1'
-      const wsUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/ws?log_id=${latestLog.id}`
+      const sseUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/sse?log_id=${latestLog.id}`

-      logSocket = new WebSocket(wsUrl)
-      logSocket.onmessage = (event) => {
-        logContent.value += event.data
+      logSource = new EventSource(sseUrl)
+      logSource.onmessage = (event) => {
+        try {
+          const data = JSON.parse(event.data)
+          logContent.value += data.text || ''
+        } catch {
+          logContent.value += event.data
+        }
+      }
+      logSource.onerror = (e) => {
+        console.error('[LogSSE] Connection error/closed', e)
+        cleanupLogSocket()
      }

      // 启动状态轮询