feat(logging): implement client log streaming and management

- Added log streaming functionality for clients, allowing real-time log access via SSE.
- Introduced LogHandler to manage log streaming requests and responses.
- Implemented LogSessionManager to handle active log sessions and listeners.
- Enhanced protocol with log-related message types and structures.
- Created Logger for client-side logging, supporting various log levels and file output.
- Developed LogViewer component for the web interface to display and filter logs.
- Updated API to support log stream creation and management.
- Added support for querying logs by level and searching through log messages.

internal/server/router/handler/interfaces.go

@@ -3,6 +3,7 @@ package handler
 import (
 	"github.com/gotunnel/internal/server/config"
 	"github.com/gotunnel/internal/server/db"
+	"github.com/gotunnel/pkg/protocol"
 )
 
 // AppInterface 应用接口
@@ -41,6 +42,9 @@ type ServerInterface interface {
 	RestartClientPlugin(clientID, pluginName, ruleName string) error
 	UpdateClientPluginConfig(clientID, pluginName, ruleName string, config map[string]string, restart bool) error
 	SendUpdateToClient(clientID, downloadURL string) error
+	// 日志流
+	StartClientLogStream(clientID, sessionID string, lines int, follow bool, level string) (<-chan protocol.LogEntry, error)
+	StopClientLogStream(sessionID string)
 }
 
 // ConfigField 配置字段

internal/server/router/handler/log.go

@@ -0,0 +1,96 @@
+package handler
+
+import (
+	"io"
+	"strconv"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+)
+
+// LogHandler 日志处理器
+type LogHandler struct {
+	app AppInterface
+}
+
+// NewLogHandler 创建日志处理器
+func NewLogHandler(app AppInterface) *LogHandler {
+	return &LogHandler{app: app}
+}
+
+// StreamLogs 流式传输客户端日志
+// @Summary 流式传输客户端日志
+// @Description 通过 Server-Sent Events 实时接收客户端日志
+// @Tags Logs
+// @Produce text/event-stream
+// @Security Bearer
+// @Param id path string true "客户端 ID"
+// @Param lines query int false "初始日志行数" default(100)
+// @Param follow query bool false "是否持续推送新日志" default(true)
+// @Param level query string false "日志级别过滤 (info, warn, error)"
+// @Success 200 {object} protocol.LogEntry
+// @Router /api/client/{id}/logs [get]
+func (h *LogHandler) StreamLogs(c *gin.Context) {
+	clientID := c.Param("id")
+
+	// 检查客户端是否在线
+	online, _, _ := h.app.GetServer().GetClientStatus(clientID)
+	if !online {
+		c.JSON(400, gin.H{"code": 400, "message": "client not online"})
+		return
+	}
+
+	// 解析查询参数
+	lines := 100
+	if v := c.Query("lines"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil {
+			lines = n
+		}
+	}
+
+	follow := true
+	if v := c.Query("follow"); v == "false" {
+		follow = false
+	}
+
+	level := c.Query("level")
+
+	// 生成会话 ID
+	sessionID := uuid.New().String()
+
+	// 启动日志流
+	logCh, err := h.app.GetServer().StartClientLogStream(clientID, sessionID, lines, follow, level)
+	if err != nil {
+		c.JSON(500, gin.H{"code": 500, "message": err.Error()})
+		return
+	}
+
+	// 设置 SSE 响应头
+	c.Writer.Header().Set("Content-Type", "text/event-stream")
+	c.Writer.Header().Set("Cache-Control", "no-cache")
+	c.Writer.Header().Set("Connection", "keep-alive")
+	c.Writer.Header().Set("X-Accel-Buffering", "no")
+
+	// 获取客户端断开信号
+	clientGone := c.Request.Context().Done()
+
+	// 流式传输日志
+	c.Stream(func(w io.Writer) bool {
+		select {
+		case <-clientGone:
+			h.app.GetServer().StopClientLogStream(sessionID)
+			return false
+		case entry, ok := <-logCh:
+			if !ok {
+				return false
+			}
+			c.SSEvent("log", entry)
+			return true
+		case <-time.After(30 * time.Second):
+			// 发送心跳
+			c.SSEvent("heartbeat", gin.H{"ts": time.Now().UnixMilli()})
+			return true
+		}
+	})
+}

internal/server/router/middleware/jwt.go

@@ -12,6 +12,14 @@ import (
 func JWTAuth(jwtAuth *auth.JWTAuth) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		authHeader := c.GetHeader("Authorization")
+
+		// 支持从 query 参数获取 token (用于 SSE 等不支持自定义 header 的场景)
+		if authHeader == "" {
+			if token := c.Query("token"); token != "" {
+				authHeader = "Bearer " + token
+			}
+		}
+
 		if authHeader == "" {
 			c.JSON(http.StatusUnauthorized, gin.H{
 				"code":    401,

internal/server/router/router.go

@@ -105,6 +105,10 @@ func (r *GinRouter) SetupRoutes(app handler.AppInterface, jwtAuth *auth.JWTAuth,
 		api.GET("/update/check/client", updateHandler.CheckClient)
 		api.POST("/update/apply/server", updateHandler.ApplyServer)
 		api.POST("/update/apply/client", updateHandler.ApplyClient)
+
+		// 日志管理
+		logHandler := handler.NewLogHandler(app)
+		api.GET("/client/:id/logs", logHandler.StreamLogs)
 	}
 }
 

internal/server/tunnel/log_session.go

@@ -0,0 +1,155 @@
+package tunnel
+
+import (
+	"net"
+	"sync"
+
+	"github.com/gotunnel/pkg/protocol"
+)
+
+// LogSessionManager 管理所有活跃的日志会话
+type LogSessionManager struct {
+	sessions map[string]*LogSession
+	mu       sync.RWMutex
+}
+
+// LogSession 日志流会话
+type LogSession struct {
+	ID        string
+	ClientID  string
+	Stream    net.Conn
+	listeners []chan protocol.LogEntry
+	mu        sync.Mutex
+	closed    bool
+}
+
+// NewLogSessionManager 创建日志会话管理器
+func NewLogSessionManager() *LogSessionManager {
+	return &LogSessionManager{
+		sessions: make(map[string]*LogSession),
+	}
+}
+
+// CreateSession 创建日志会话
+func (m *LogSessionManager) CreateSession(clientID, sessionID string, stream net.Conn) *LogSession {
+	session := &LogSession{
+		ID:        sessionID,
+		ClientID:  clientID,
+		Stream:    stream,
+		listeners: make([]chan protocol.LogEntry, 0),
+	}
+
+	m.mu.Lock()
+	m.sessions[sessionID] = session
+	m.mu.Unlock()
+
+	return session
+}
+
+// GetSession 获取会话
+func (m *LogSessionManager) GetSession(sessionID string) *LogSession {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.sessions[sessionID]
+}
+
+// RemoveSession 移除会话
+func (m *LogSessionManager) RemoveSession(sessionID string) {
+	m.mu.Lock()
+	if session, ok := m.sessions[sessionID]; ok {
+		session.Close()
+		delete(m.sessions, sessionID)
+	}
+	m.mu.Unlock()
+}
+
+// GetSessionsByClient 获取客户端的所有会话
+func (m *LogSessionManager) GetSessionsByClient(clientID string) []*LogSession {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	var sessions []*LogSession
+	for _, session := range m.sessions {
+		if session.ClientID == clientID {
+			sessions = append(sessions, session)
+		}
+	}
+	return sessions
+}
+
+// CleanupClientSessions 清理客户端的所有会话
+func (m *LogSessionManager) CleanupClientSessions(clientID string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	for id, session := range m.sessions {
+		if session.ClientID == clientID {
+			session.Close()
+			delete(m.sessions, id)
+		}
+	}
+}
+
+// AddListener 添加监听器
+func (s *LogSession) AddListener() <-chan protocol.LogEntry {
+	ch := make(chan protocol.LogEntry, 100)
+	s.mu.Lock()
+	s.listeners = append(s.listeners, ch)
+	s.mu.Unlock()
+	return ch
+}
+
+// RemoveListener 移除监听器
+func (s *LogSession) RemoveListener(ch <-chan protocol.LogEntry) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for i, listener := range s.listeners {
+		if listener == ch {
+			close(listener)
+			s.listeners = append(s.listeners[:i], s.listeners[i+1:]...)
+			break
+		}
+	}
+}
+
+// Broadcast 广播日志条目到所有监听器
+func (s *LogSession) Broadcast(entry protocol.LogEntry) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for _, ch := range s.listeners {
+		select {
+		case ch <- entry:
+		default:
+			// 监听器太慢，丢弃日志
+		}
+	}
+}
+
+// Close 关闭会话
+func (s *LogSession) Close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.closed {
+		return
+	}
+	s.closed = true
+
+	for _, ch := range s.listeners {
+		close(ch)
+	}
+	s.listeners = nil
+
+	if s.Stream != nil {
+		s.Stream.Close()
+	}
+}
+
+// IsClosed 检查会话是否已关闭
+func (s *LogSession) IsClosed() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.closed
+}

internal/server/tunnel/server.go

@@ -65,6 +65,7 @@ type Server struct {
 	listener       net.Listener    // 主监听器
 	shutdown       chan struct{}   // 关闭信号
 	wg             sync.WaitGroup  // 等待所有连接关闭
+	logSessions    *LogSessionManager // 日志会话管理器
 }
 
 // JSPluginEntry JS 插件条目
@@ -102,6 +103,7 @@ func NewServer(cs db.ClientStore, bindAddr string, bindPort int, token string, h
 		clients:     make(map[string]*ClientSession),
 		connSem:     make(chan struct{}, maxConnections),
 		shutdown:    make(chan struct{}),
+		logSessions: NewLogSessionManager(),
 	}
 }
 
@@ -1474,3 +1476,96 @@ func (s *Server) SendUpdateToClient(clientID, downloadURL string) error {
 	log.Printf("[Server] Update command sent to client %s: %s", clientID, downloadURL)
 	return nil
 }
+
+// StartClientLogStream 启动客户端日志流
+func (s *Server) StartClientLogStream(clientID, sessionID string, lines int, follow bool, level string) (<-chan protocol.LogEntry, error) {
+	s.mu.RLock()
+	cs, ok := s.clients[clientID]
+	s.mu.RUnlock()
+
+	if !ok {
+		return nil, fmt.Errorf("client %s not found or not online", clientID)
+	}
+
+	// 打开到客户端的流
+	stream, err := cs.Session.Open()
+	if err != nil {
+		return nil, err
+	}
+
+	// 发送日志请求
+	req := protocol.LogRequest{
+		SessionID: sessionID,
+		Lines:     lines,
+		Follow:    follow,
+		Level:     level,
+	}
+	msg, _ := protocol.NewMessage(protocol.MsgTypeLogRequest, req)
+	if err := protocol.WriteMessage(stream, msg); err != nil {
+		stream.Close()
+		return nil, err
+	}
+
+	// 创建会话
+	session := s.logSessions.CreateSession(clientID, sessionID, stream)
+	listener := session.AddListener()
+
+	// 启动 goroutine 读取客户端日志
+	go s.readClientLogs(session, stream)
+
+	return listener, nil
+}
+
+// readClientLogs 读取客户端日志并广播到监听器
+func (s *Server) readClientLogs(session *LogSession, stream net.Conn) {
+	defer s.logSessions.RemoveSession(session.ID)
+
+	for {
+		msg, err := protocol.ReadMessage(stream)
+		if err != nil {
+			return
+		}
+
+		if msg.Type != protocol.MsgTypeLogData {
+			continue
+		}
+
+		var data protocol.LogData
+		if err := msg.ParsePayload(&data); err != nil {
+			continue
+		}
+
+		for _, entry := range data.Entries {
+			session.Broadcast(entry)
+		}
+
+		if data.EOF {
+			return
+		}
+	}
+}
+
+// StopClientLogStream 停止客户端日志流
+func (s *Server) StopClientLogStream(sessionID string) {
+	session := s.logSessions.GetSession(sessionID)
+	if session == nil {
+		return
+	}
+
+	// 发送停止请求到客户端
+	s.mu.RLock()
+	cs, ok := s.clients[session.ClientID]
+	s.mu.RUnlock()
+
+	if ok {
+		stream, err := cs.Session.Open()
+		if err == nil {
+			req := protocol.LogStopRequest{SessionID: sessionID}
+			msg, _ := protocol.NewMessage(protocol.MsgTypeLogStop, req)
+			protocol.WriteMessage(stream, msg)
+			stream.Close()
+		}
+	}
+
+	s.logSessions.RemoveSession(sessionID)
+}