feat: 添加SSE实时更新、键盘导航和文件系统优化

主要更新:
- 实现 SSE (Server-Sent Events) 实时文件系统监听
- 添加完整的键盘导航功能和快捷键支持
- 优化文件系统 API 集成和错误处理
- 添加优雅关机和连接管理
- 新增详细的功能文档和调试指南

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

src/rear/go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/bytedance/sonic/loader v0.2.4 // indirect
 	github.com/cloudwego/base64x v0.1.5 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.9 // indirect
 	github.com/gin-contrib/sse v1.1.0 // indirect
 	github.com/glebarez/go-sqlite v1.21.2 // indirect

src/rear/go.sum

@@ -13,6 +13,10 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
+github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
+github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
+github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/gabriel-vasile/mimetype v1.4.9 h1:5k+WDwEsD9eTLL8Tz3L0VnmVh9QxGjRmjBvAG7U/oYY=
 github.com/gabriel-vasile/mimetype v1.4.9/go.mod h1:WnSQhFKJuBlRyLiKohA/2DtIlPFAbguNaG7QCHcyGok=
 github.com/gin-contrib/cors v1.7.5 h1:cXC9SmofOrRg0w9PigwGlHG3ztswH6bqq4vJVXnvYMk=
@@ -106,6 +110,7 @@ golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM=
 golang.org/x/crypto v0.39.0/go.mod h1:L+Xg3Wf6HoL4Bn4238Z6ft6KfEpN0tJGo53AAPC632U=
 golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw=
 golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA=
+golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
 golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=

src/rear/internal/config/loader.go

@@ -122,8 +122,8 @@ func GetDefaultFileConfig() *FileConfig {
 			Port:         getEnv("PORT", "8080"),
 			Mode:         getEnv("GIN_MODE", "debug"),
 			ReadTimeout:  30 * time.Second,
-			WriteTimeout: 30 * time.Second,
-			IdleTimeout:  60 * time.Second,
+			WriteTimeout: 0, // SSE 需要长连接，设置为 0（无限期）
+			IdleTimeout:  10 * time.Minute, // 空闲超时设置为 10 分钟
 			Development:  nil, // 默认不设置，表示生产阶段
 		},
 		Tools: ToolsConfig{

src/rear/internal/db/db.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 	"gorm.io/driver/mysql"
-	"gorm.io/driver/sqlite"
+	sqlite "github.com/glebarez/sqlite"
 	"gorm.io/gorm/logger"
 	"rear/internal/config"
 	"sync"

src/rear/internal/handler/filesystem_handler.go

@@ -1,10 +1,13 @@
 package handler
 
 import (
+	"encoding/json"
 	"net/http"
 	"rear/internal/model"
 	"rear/internal/service"
+	"rear/internal/utils"
 	"rear/pkg/logger"
+	"rear/pkg/sse"
 	"strconv"
 
 	"github.com/gin-gonic/gin"
@@ -14,28 +17,131 @@ import (
 // FileSystemHandler 文件系统处理器
 type FileSystemHandler struct {
 	fileSystemService *service.FileSystemService
+	fileWatcher       *utils.FileSystemWatcher
+	sseManager        *sse.Manager
+	eventThrottle     chan struct{} // 事件节流通道
 }
 
 // NewFileSystemHandler 创建文件系统处理器实例
-func NewFileSystemHandler() *FileSystemHandler {
+func NewFileSystemHandler(sseManager *sse.Manager) *FileSystemHandler {
+	// 创建事件节流通道（最多缓存10个待发送事件）
+	eventThrottle := make(chan struct{}, 10)
+
+	// 创建文件系统监听器，事件回调通过 SSE 推送
+	fileWatcher, err := utils.NewFileSystemWatcher(func(event *utils.FileSystemEvent) {
+		// 防御性编程：捕获回调中的panic
+		defer func() {
+			if r := recover(); r != nil {
+				logger.Error("文件系统事件回调panic",
+					zap.Any("panic", r),
+					zap.String("type", event.Type),
+					zap.String("path", event.Path))
+			}
+		}()
+
+		// 验证事件有效性
+		if event == nil {
+			logger.Warn("收到空的文件系统事件")
+			return
+		}
+
+		// 将事件转换为 JSON
+		eventData, err := json.Marshal(event)
+		if err != nil {
+			logger.Error("序列化文件系统事件失败",
+				zap.Error(err),
+				zap.String("type", event.Type),
+				zap.String("path", event.Path))
+			return
+		}
+
+		// 验证JSON字符串的有效性
+		eventString := string(eventData)
+		if len(eventString) == 0 {
+			logger.Error("序列化后的JSON为空",
+				zap.String("type", event.Type),
+				zap.String("path", event.Path))
+			return
+		}
+
+		// 检查是否包含非法控制字符
+		for i, r := range eventString {
+			if r < 32 && r != '\n' && r != '\r' && r != '\t' {
+				logger.Error("JSON包含非法控制字符",
+					zap.Int("position", i),
+					zap.Int32("char", r),
+					zap.String("json", eventString))
+				return
+			}
+		}
+
+		// 非阻塞发送：如果节流通道已满，丢弃事件（避免阻塞监听器）
+		select {
+		case eventThrottle <- struct{}{}:
+			// 异步广播事件，避免阻塞 fsnotify 循环
+			go func(eventStr string, ev *utils.FileSystemEvent) {
+				defer func() {
+					<-eventThrottle // 释放节流槽位
+					if r := recover(); r != nil {
+						logger.Error("SSE广播panic", zap.Any("panic", r))
+					}
+				}()
+
+				// 使用新的订阅广播方法：只向订阅了该路径的客户端发送
+				// 使用事件的 WatchedPath 字段作为订阅路径
+				sseManager.BroadcastToSubscribers(ev.WatchedPath, "filesystem-change", eventStr)
+
+				// 🟢 关键日志: 记录每次文件系统变化推送
+				logger.Info("📤 文件系统变化事件已推送",
+					zap.String("event_type", ev.Type),
+					zap.String("file_path", ev.Path),
+					zap.String("file_name", ev.Name),
+					zap.String("watched_path", ev.WatchedPath),
+					zap.Bool("is_dir", ev.IsDir),
+					zap.String("timestamp", ev.Timestamp.Format("2006-01-02 15:04:05")))
+			}(eventString, event)
+		default:
+			// 节流通道已满，丢弃事件
+			logger.Warn("文件系统事件频率过高，跳过事件",
+				zap.String("type", event.Type),
+				zap.String("path", event.Path))
+		}
+	})
+
+	if err != nil {
+		logger.Error("创建文件系统监听器失败", zap.Error(err))
+		return nil
+	}
+
+	// 启动监听器
+	fileWatcher.Start()
+
 	return &FileSystemHandler{
 		fileSystemService: service.NewFileSystemService(),
+		fileWatcher:       fileWatcher,
+		sseManager:        sseManager,
+		eventThrottle:     eventThrottle,
 	}
 }
 
+// GetFileWatcher 获取文件监听器（用于清理）
+func (h *FileSystemHandler) GetFileWatcher() *utils.FileSystemWatcher {
+	return h.fileWatcher
+}
+
 // BrowseFileSystem 浏览文件系统
 // GET /api/v1/filesystem/browse?path=/path/to/dir
 func (h *FileSystemHandler) BrowseFileSystem(c *gin.Context) {
 	path := c.Query("path")
 
-	logger.Info("Browse file system request",
+	logger.Info("🔄 浏览文件系统请求",
 		zap.String("path", path),
 		zap.String("client_ip", c.ClientIP()),
 	)
 
 	result, err := h.fileSystemService.Browse(path)
 	if err != nil {
-		logger.Error("Failed to browse file system",
+		logger.Error("❌ 浏览文件系统失败",
 			zap.String("path", path),
 			zap.Error(err),
 		)
@@ -46,6 +152,11 @@ func (h *FileSystemHandler) BrowseFileSystem(c *gin.Context) {
 		return
 	}
 
+	logger.Info("✅ 浏览文件系统成功",
+		zap.String("current_path", result.CurrentPath),
+		zap.Int("items_count", len(result.Items)),
+	)
+
 	c.JSON(http.StatusOK, model.Response{
 		Code:    http.StatusOK,
 		Message: "获取文件系统信息成功",
@@ -378,4 +489,121 @@ func (h *FileSystemHandler) SearchFiles(c *gin.Context) {
 		Message: "搜索完成",
 		Data:    result,
 	})
+}
+
+// WatchPath 订阅文件夹监听
+// POST /api/v1/filesystem/watch
+// Body: {"path": "/path/to/watch"}
+func (h *FileSystemHandler) WatchPath(c *gin.Context) {
+	var req struct {
+		Path string `json:"path" binding:"required"`
+	}
+
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, model.Response{
+			Code:    http.StatusBadRequest,
+			Message: "请求参数错误: " + err.Error(),
+		})
+		return
+	}
+
+	logger.Info("👀 订阅文件夹监听请求",
+		zap.String("path", req.Path),
+		zap.String("client_ip", c.ClientIP()),
+	)
+
+	// 添加监听
+	err := h.fileWatcher.Watch(req.Path)
+	if err != nil {
+		logger.Error("❌ 添加监听失败",
+			zap.String("path", req.Path),
+			zap.Error(err),
+		)
+		c.JSON(http.StatusInternalServerError, model.Response{
+			Code:    http.StatusInternalServerError,
+			Message: "添加监听失败: " + err.Error(),
+		})
+		return
+	}
+
+	watchedPaths := h.fileWatcher.GetWatchedPaths()
+	logger.Info("✅ 已开始监听文件夹",
+		zap.String("path", req.Path),
+		zap.Int("total_watched", len(watchedPaths)),
+	)
+
+	c.JSON(http.StatusOK, model.Response{
+		Code:    http.StatusOK,
+		Message: "已开始监听文件夹",
+		Data: map[string]interface{}{
+			"path":          req.Path,
+			"watched_paths": watchedPaths,
+		},
+	})
+}
+
+// UnwatchPath 取消文件夹监听
+// POST /api/v1/filesystem/unwatch
+// Body: {"path": "/path/to/unwatch"}
+func (h *FileSystemHandler) UnwatchPath(c *gin.Context) {
+	var req struct {
+		Path string `json:"path" binding:"required"`
+	}
+
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, model.Response{
+			Code:    http.StatusBadRequest,
+			Message: "请求参数错误: " + err.Error(),
+		})
+		return
+	}
+
+	logger.Info("🚫 取消文件夹监听请求",
+		zap.String("path", req.Path),
+		zap.String("client_ip", c.ClientIP()),
+	)
+
+	// 移除监听
+	err := h.fileWatcher.Unwatch(req.Path)
+	if err != nil {
+		logger.Error("❌ 移除监听失败",
+			zap.String("path", req.Path),
+			zap.Error(err),
+		)
+		c.JSON(http.StatusInternalServerError, model.Response{
+			Code:    http.StatusInternalServerError,
+			Message: "移除监听失败: " + err.Error(),
+		})
+		return
+	}
+
+	watchedPaths := h.fileWatcher.GetWatchedPaths()
+	logger.Info("✅ 已停止监听文件夹",
+		zap.String("path", req.Path),
+		zap.Int("total_watched", len(watchedPaths)),
+	)
+
+	c.JSON(http.StatusOK, model.Response{
+		Code:    http.StatusOK,
+		Message: "已停止监听文件夹",
+		Data: map[string]interface{}{
+			"path":          req.Path,
+			"watched_paths": watchedPaths,
+		},
+	})
+}
+
+// GetWatchedPaths 获取当前所有监听的路径
+// GET /api/v1/filesystem/watched
+func (h *FileSystemHandler) GetWatchedPaths(c *gin.Context) {
+	paths := h.fileWatcher.GetWatchedPaths()
+
+	c.JSON(http.StatusOK, model.Response{
+		Code:    http.StatusOK,
+		Message: "获取监听路径成功",
+		Data: map[string]interface{}{
+			"count": len(paths),
+			"paths": paths,
+		},
+	})
 }

src/rear/internal/handler/sse_handler.go

@@ -13,7 +13,7 @@ import (
 )
 
 type SSEHandler struct {
-	manager *sse.Manager
+	Manager *sse.Manager // 公开字段以便其他 Handler 访问
 }
 
 type SSEResponse struct {
@@ -32,19 +32,26 @@ type ClientInfo struct {
 
 func NewSSEHandler() *SSEHandler {
 	opts := sse.DefaultOptions()
-	opts.PingInterval = 30 * time.Second
+	opts.PingInterval = 10 * time.Second  // 10秒心跳，配合 keepalive 的 5秒
 	opts.MaxClients = 500
 	opts.EventBufferSize = 200
-	
+
 	manager := sse.NewManager(opts)
-	
+
 	return &SSEHandler{
-		manager: manager,
+		Manager: manager,
 	}
 }
 
 func (h *SSEHandler) HandleSSEConnection(c *gin.Context) {
-	h.manager.RegisterClient(c)
+	// 添加日志确认方法被调用
+	fmt.Printf("==== SSE HandleSSEConnection called, client IP: %s ====\n", c.ClientIP())
+
+	// 对于 SSE 连接，需要设置无限期的超时
+	// 通过设置 deadline 为很远的未来时间来绕过 WriteTimeout
+	c.Request.Context()
+
+	h.Manager.RegisterClient(c)
 }
 
 func (h *SSEHandler) BroadcastMessage(c *gin.Context) {
@@ -62,16 +69,16 @@ func (h *SSEHandler) BroadcastMessage(c *gin.Context) {
 	}
 	
 	if request.EventType != "" {
-		h.manager.BroadcastEvent(request.EventType, request.Message)
+		h.Manager.BroadcastEvent(request.EventType, request.Message)
 	} else {
-		h.manager.Broadcast(request.Message)
+		h.Manager.Broadcast(request.Message)
 	}
 	
 	c.JSON(http.StatusOK, SSEResponse{
 		Success: true,
 		Message: "Message broadcasted successfully",
 		Data: map[string]interface{}{
-			"client_count": h.manager.GetClientCount(),
+			"client_count": h.Manager.GetClientCount(),
 			"message":      request.Message,
 			"event_type":   request.EventType,
 		},
@@ -100,7 +107,7 @@ func (h *SSEHandler) SendToClient(c *gin.Context) {
 		return
 	}
 	
-	success := h.manager.SendToClient(clientID, request.Message)
+	success := h.Manager.SendToClient(clientID, request.Message)
 	if !success {
 		c.JSON(http.StatusNotFound, SSEResponse{
 			Success: false,
@@ -120,11 +127,11 @@ func (h *SSEHandler) SendToClient(c *gin.Context) {
 }
 
 func (h *SSEHandler) GetClients(c *gin.Context) {
-	clientIDs := h.manager.GetClientIDs()
+	clientIDs := h.Manager.GetClientIDs()
 	clients := make([]ClientInfo, 0, len(clientIDs))
 	
 	for _, id := range clientIDs {
-		if client, exists := h.manager.GetClientInfo(id); exists {
+		if client, exists := h.Manager.GetClientInfo(id); exists {
 			clients = append(clients, ClientInfo{
 				ID:          client.ID,
 				RemoteAddr:  client.RemoteAddr,
@@ -155,7 +162,7 @@ func (h *SSEHandler) GetClientInfo(c *gin.Context) {
 		return
 	}
 	
-	client, exists := h.manager.GetClientInfo(clientID)
+	client, exists := h.Manager.GetClientInfo(clientID)
 	if !exists {
 		c.JSON(http.StatusNotFound, SSEResponse{
 			Success: false,
@@ -189,7 +196,7 @@ func (h *SSEHandler) DisconnectClient(c *gin.Context) {
 		return
 	}
 	
-	h.manager.UnregisterClient(clientID)
+	h.Manager.UnregisterClient(clientID)
 	
 	c.JSON(http.StatusOK, SSEResponse{
 		Success: true,
@@ -202,8 +209,8 @@ func (h *SSEHandler) DisconnectClient(c *gin.Context) {
 
 func (h *SSEHandler) GetStats(c *gin.Context) {
 	stats := map[string]interface{}{
-		"total_clients":   h.manager.GetClientCount(),
-		"client_ids":      h.manager.GetClientIDs(),
+		"total_clients":   h.Manager.GetClientCount(),
+		"client_ids":      h.Manager.GetClientIDs(),
 		"server_time":     time.Now().Format(time.RFC3339),
 		"uptime_seconds":  time.Since(time.Now().Add(-time.Hour)).Seconds(), // 示例值
 	}
@@ -239,7 +246,7 @@ func (h *SSEHandler) TestEvent(c *gin.Context) {
 			}
 			
 			messageJSON, _ := json.Marshal(message)
-			h.manager.BroadcastEvent(eventType, string(messageJSON))
+			h.Manager.BroadcastEvent(eventType, string(messageJSON))
 			
 			time.Sleep(time.Duration(intervalDur) * time.Second)
 		}
@@ -257,11 +264,115 @@ func (h *SSEHandler) TestEvent(c *gin.Context) {
 }
 
 func (h *SSEHandler) GetEventChannel() chan<- string {
-	return h.manager.GetEventChannel()
+	return h.Manager.GetEventChannel()
 }
 
 func (h *SSEHandler) Close() {
-	if h.manager != nil {
-		h.manager.Close()
+	if h.Manager != nil {
+		h.Manager.Close()
 	}
+}
+
+// ============ 订阅管理接口 ============
+
+// Subscribe 客户端订阅指定路径
+// POST /api/v1/sse/subscribe
+func (h *SSEHandler) Subscribe(c *gin.Context) {
+	var request struct {
+		ClientID string `json:"client_id" binding:"required"`
+		Path     string `json:"path" binding:"required"`
+	}
+
+	if err := c.ShouldBindJSON(&request); err != nil {
+		c.JSON(http.StatusBadRequest, SSEResponse{
+			Success: false,
+			Message: "Invalid request format: " + err.Error(),
+		})
+		return
+	}
+
+	err := h.Manager.Subscribe(request.ClientID, request.Path)
+	if err != nil {
+		c.JSON(http.StatusNotFound, SSEResponse{
+			Success: false,
+			Message: err.Error(),
+		})
+		return
+	}
+
+	c.JSON(http.StatusOK, SSEResponse{
+		Success: true,
+		Message: "Subscribed successfully",
+		Data: map[string]interface{}{
+			"client_id": request.ClientID,
+			"path":      request.Path,
+		},
+	})
+}
+
+// Unsubscribe 客户端取消订阅指定路径
+// POST /api/v1/sse/unsubscribe
+func (h *SSEHandler) Unsubscribe(c *gin.Context) {
+	var request struct {
+		ClientID string `json:"client_id" binding:"required"`
+		Path     string `json:"path" binding:"required"`
+	}
+
+	if err := c.ShouldBindJSON(&request); err != nil {
+		c.JSON(http.StatusBadRequest, SSEResponse{
+			Success: false,
+			Message: "Invalid request format: " + err.Error(),
+		})
+		return
+	}
+
+	err := h.Manager.Unsubscribe(request.ClientID, request.Path)
+	if err != nil {
+		c.JSON(http.StatusNotFound, SSEResponse{
+			Success: false,
+			Message: err.Error(),
+		})
+		return
+	}
+
+	c.JSON(http.StatusOK, SSEResponse{
+		Success: true,
+		Message: "Unsubscribed successfully",
+		Data: map[string]interface{}{
+			"client_id": request.ClientID,
+			"path":      request.Path,
+		},
+	})
+}
+
+// GetClientSubscriptions 获取客户端的所有订阅
+// GET /api/v1/sse/subscriptions/:clientId
+func (h *SSEHandler) GetClientSubscriptions(c *gin.Context) {
+	clientID := c.Param("clientId")
+	if clientID == "" {
+		c.JSON(http.StatusBadRequest, SSEResponse{
+			Success: false,
+			Message: "Client ID is required",
+		})
+		return
+	}
+
+	subscriptions, err := h.Manager.GetSubscriptions(clientID)
+	if err != nil {
+		c.JSON(http.StatusNotFound, SSEResponse{
+			Success: false,
+			Message: err.Error(),
+		})
+		return
+	}
+
+	c.JSON(http.StatusOK, SSEResponse{
+		Success: true,
+		Message: "Subscriptions retrieved successfully",
+		Data: map[string]interface{}{
+			"client_id":     clientID,
+			"subscriptions": subscriptions,
+			"count":         len(subscriptions),
+		},
+	})
 }

src/rear/internal/router/router.go

@@ -7,8 +7,12 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
+// CleanupFunc 清理函数类型
+type CleanupFunc func()
+
 // SetupRoutes 设置路由
-func SetupRoutes(r *gin.Engine, contain *container.DbContainer, imgContain *container.TaskContainer) {
+// 返回一个清理函数，用于在服务器关闭时清理资源
+func SetupRoutes(r *gin.Engine, contain *container.DbContainer, imgContain *container.TaskContainer) CleanupFunc {
 	// 默认访问
 	r.GET("/", handler.BasicResponse)
 
@@ -21,8 +25,11 @@ func SetupRoutes(r *gin.Engine, contain *container.DbContainer, imgContain *cont
 	exifHandler := handler.NewExifHandler(contain)
 	photoHandler := handler.NewPhotoHandler(contain, imgContain)
 
-	// 文件系统处理
-	fileSystemHandler := handler.NewFileSystemHandler()
+	// SSE 实时通信处理
+	sseHandler := handler.NewSSEHandler()
+
+	// 文件系统处理 (需要 SSE Manager)
+	fileSystemHandler := handler.NewFileSystemHandler(sseHandler.Manager)
 	fileOperationsHandler := handler.NewFileOperationsHandler()
 
 	// API版本组
@@ -82,6 +89,24 @@ func SetupRoutes(r *gin.Engine, contain *container.DbContainer, imgContain *cont
 			assets.GET("/:hash", photoHandler.GetPhotoAssets) // 获取图像详细信息
 		}
 
+		// SSE 实时通信路由
+		sse := v1.Group("/sse")
+		{
+			sse.GET("/connect", sseHandler.HandleSSEConnection)            // SSE 连接
+			sse.POST("/broadcast", sseHandler.BroadcastMessage)            // 广播消息
+			sse.POST("/send/:clientId", sseHandler.SendToClient)           // 发送给特定客户端
+			sse.GET("/clients", sseHandler.GetClients)                     // 获取所有客户端
+			sse.GET("/clients/:clientId", sseHandler.GetClientInfo)        // 获取客户端信息
+			sse.DELETE("/clients/:clientId", sseHandler.DisconnectClient)  // 断开客户端
+			sse.GET("/stats", sseHandler.GetStats)                         // 获取统计信息
+			sse.GET("/test", sseHandler.TestEvent)                         // 测试事件
+
+			// 订阅管理路由
+			sse.POST("/subscribe", sseHandler.Subscribe)                          // 订阅路径
+			sse.POST("/unsubscribe", sseHandler.Unsubscribe)                      // 取消订阅
+			sse.GET("/subscriptions/:clientId", sseHandler.GetClientSubscriptions) // 获取订阅列表
+		}
+
 		// 文件系统相关路由
 		filesystem := v1.Group("/filesystem")
 		{
@@ -95,6 +120,11 @@ func SetupRoutes(r *gin.Engine, contain *container.DbContainer, imgContain *cont
 			filesystem.DELETE("/item", fileOperationsHandler.DeleteItem)           // 删除文件或目录
 			filesystem.PUT("/item/move", fileOperationsHandler.MoveItem)           // 移动/重命名
 			filesystem.POST("/item/copy", fileOperationsHandler.CopyItem)          // 复制文件或目录
+
+			// 文件系统监听相关路由
+			filesystem.POST("/watch", fileSystemHandler.WatchPath)                 // 订阅文件夹监听
+			filesystem.POST("/unwatch", fileSystemHandler.UnwatchPath)             // 取消文件夹监听
+			filesystem.GET("/watched", fileSystemHandler.GetWatchedPaths)          // 获取监听路径列表
 		}
 	}
 	// 开发组
@@ -103,4 +133,17 @@ func SetupRoutes(r *gin.Engine, contain *container.DbContainer, imgContain *cont
 		// 处理图片格式
 		dev.GET("/tool/exiftool/get_exif", devImageHandler.GetExif)
 	}
+
+	// 返回清理函数
+	return func() {
+		// 关闭 SSE Manager
+		if sseHandler != nil && sseHandler.Manager != nil {
+			sseHandler.Manager.Close()
+		}
+
+		// 关闭文件系统监听器
+		if fileSystemHandler != nil && fileSystemHandler.GetFileWatcher() != nil {
+			fileSystemHandler.GetFileWatcher().Close()
+		}
+	}
 }

src/rear/internal/service/filesystem_service.go

@@ -99,8 +99,19 @@ func (fs *FileSystemService) Browse(path string) (*FileSystemResponse, error) {
 		return fs.getRootLevel()
 	}
 
-	// 清理路径
-	path = filepath.Clean(path)
+	// 清理路径，但需要特殊处理 Windows 驱动器根路径
+	// filepath.Clean("D:") 会变成 "D:." 导致前端面包屑出现问题
+	cleanPath := filepath.Clean(path)
+
+	// Windows 驱动器根路径修正：将 "D:." 转换为 "D:\"
+	if runtime.GOOS == "windows" {
+		// 匹配 "C:.", "D:." 等模式
+		if len(cleanPath) == 3 && cleanPath[1] == ':' && cleanPath[2] == '.' {
+			cleanPath = cleanPath[:2] + "\\"
+		}
+	}
+
+	path = cleanPath
 
 	// 检查路径是否存在
 	if !utils.FileUtils.Exists(path) {

src/rear/internal/utils/filesystem_watcher.go

@@ -0,0 +1,289 @@
+package utils
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+// FileSystemEvent 文件系统变化事件
+type FileSystemEvent struct {
+	Type        string    `json:"type"`         // create, modify, delete, rename
+	Path        string    `json:"path"`         // 变化的文件/文件夹路径
+	Name        string    `json:"name"`         // 文件/文件夹名称
+	Timestamp   time.Time `json:"timestamp"`    // 事件时间
+	IsDir       bool      `json:"is_dir"`       // 是否为目录
+	WatchedPath string    `json:"watched_path"` // 被监听的根路径（用于订阅匹配）
+}
+
+// FileSystemWatcher 文件系统监听器
+type FileSystemWatcher struct {
+	watcher       *fsnotify.Watcher
+	watchedPaths  map[string]int // 路径 -> 订阅数
+	pathsMutex    sync.RWMutex
+	eventCallback func(*FileSystemEvent) // 事件回调
+	stopChan      chan struct{}
+	running       bool
+	runMutex      sync.Mutex
+}
+
+// NewFileSystemWatcher 创建文件系统监听器
+func NewFileSystemWatcher(eventCallback func(*FileSystemEvent)) (*FileSystemWatcher, error) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("创建文件系统监听器失败: %w", err)
+	}
+
+	fsw := &FileSystemWatcher{
+		watcher:       watcher,
+		watchedPaths:  make(map[string]int),
+		eventCallback: eventCallback,
+		stopChan:      make(chan struct{}),
+		running:       false,
+	}
+
+	return fsw, nil
+}
+
+// Start 启动监听
+func (fsw *FileSystemWatcher) Start() {
+	fsw.runMutex.Lock()
+	if fsw.running {
+		fsw.runMutex.Unlock()
+		return
+	}
+	fsw.running = true
+	fsw.runMutex.Unlock()
+
+	go fsw.watchLoop()
+}
+
+// watchLoop 监听循环
+func (fsw *FileSystemWatcher) watchLoop() {
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Printf("文件系统监听循环 panic: %v\n", r)
+		}
+	}()
+
+	for {
+		select {
+		case <-fsw.stopChan:
+			return
+
+		case event, ok := <-fsw.watcher.Events:
+			if !ok {
+				return
+			}
+			fsw.handleEvent(event)
+
+		case err, ok := <-fsw.watcher.Errors:
+			if !ok {
+				return
+			}
+			fmt.Printf("文件系统监听错误: %v\n", err)
+		}
+	}
+}
+
+// handleEvent 处理文件系统事件
+func (fsw *FileSystemWatcher) handleEvent(event fsnotify.Event) {
+	// 防御性编程：捕获 panic
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Printf("文件系统事件处理 panic: %v, 事件: %+v\n", r, event)
+		}
+	}()
+
+	if fsw.eventCallback == nil {
+		return
+	}
+
+	// 查找事件路径对应的被监听根路径
+	watchedPath := fsw.findWatchedPath(event.Name)
+
+	// 将 fsnotify.Event 转换为自定义事件
+	fsEvent := &FileSystemEvent{
+		Path:        event.Name,
+		Name:        filepath.Base(event.Name),
+		Timestamp:   time.Now(),
+		IsDir:       false, // 默认为文件
+		WatchedPath: watchedPath,
+	}
+
+	// 判断事件类型
+	switch {
+	case event.Op&fsnotify.Create == fsnotify.Create:
+		fsEvent.Type = "create"
+	case event.Op&fsnotify.Write == fsnotify.Write:
+		fsEvent.Type = "modify"
+	case event.Op&fsnotify.Remove == fsnotify.Remove:
+		fsEvent.Type = "delete"
+	case event.Op&fsnotify.Rename == fsnotify.Rename:
+		fsEvent.Type = "rename"
+	case event.Op&fsnotify.Chmod == fsnotify.Chmod:
+		// 忽略权限变化
+		return
+	default:
+		// 未知事件类型，记录并忽略
+		fmt.Printf("未知文件系统事件类型: %v, 路径: %s\n", event.Op, event.Name)
+		return
+	}
+
+	// 判断是否为目录
+	// 注意：删除和重命名事件时文件可能已不存在，需要安全处理
+	if fsEvent.Type != "delete" && fsEvent.Type != "rename" {
+		if info, err := os.Stat(event.Name); err == nil {
+			fsEvent.IsDir = info.IsDir()
+		}
+	}
+
+	// 调用回调函数
+	fsw.eventCallback(fsEvent)
+}
+
+// findWatchedPath 查找事件路径对应的被监听根路径
+func (fsw *FileSystemWatcher) findWatchedPath(eventPath string) string {
+	fsw.pathsMutex.RLock()
+	defer fsw.pathsMutex.RUnlock()
+
+	// 清理事件路径
+	eventPath = filepath.Clean(eventPath)
+
+	// 遍历所有被监听的路径，找到最长匹配的根路径
+	var bestMatch string
+	maxLen := 0
+
+	for watchedPath := range fsw.watchedPaths {
+		cleanWatched := filepath.Clean(watchedPath)
+
+		// 检查事件路径是否在该监听路径下
+		rel, err := filepath.Rel(cleanWatched, eventPath)
+		if err == nil && !filepath.IsAbs(rel) {
+			// 找到匹配的监听路径，选择最长的（最具体的）
+			if len(cleanWatched) > maxLen {
+				bestMatch = watchedPath
+				maxLen = len(cleanWatched)
+			}
+		}
+	}
+
+	return bestMatch
+}
+
+// Watch 添加监听路径
+func (fsw *FileSystemWatcher) Watch(path string) error {
+	fsw.pathsMutex.Lock()
+	defer fsw.pathsMutex.Unlock()
+
+	// 规范化路径
+	absPath, err := filepath.Abs(path)
+	if err != nil {
+		return fmt.Errorf("获取绝对路径失败: %w", err)
+	}
+
+	// 如果已经在监听，增加订阅计数
+	if count, exists := fsw.watchedPaths[absPath]; exists {
+		fsw.watchedPaths[absPath] = count + 1
+		fmt.Printf("文件夹监听订阅数增加: %s (订阅数: %d)\n", absPath, count+1)
+		return nil
+	}
+
+	// 添加到 fsnotify 监听
+	if err := fsw.watcher.Add(absPath); err != nil {
+		return fmt.Errorf("添加监听失败: %w", err)
+	}
+
+	fsw.watchedPaths[absPath] = 1
+	fmt.Printf("开始监听文件夹: %s\n", absPath)
+
+	return nil
+}
+
+// Unwatch 移除监听路径
+func (fsw *FileSystemWatcher) Unwatch(path string) error {
+	fsw.pathsMutex.Lock()
+	defer fsw.pathsMutex.Unlock()
+
+	// 检查监听器是否已经停止
+	fsw.runMutex.Lock()
+	isRunning := fsw.running
+	fsw.runMutex.Unlock()
+
+	// 规范化路径
+	absPath, err := filepath.Abs(path)
+	if err != nil {
+		return fmt.Errorf("获取绝对路径失败: %w", err)
+	}
+
+	count, exists := fsw.watchedPaths[absPath]
+	if !exists {
+		return nil // 路径未被监听
+	}
+
+	// 减少订阅计数
+	if count > 1 {
+		fsw.watchedPaths[absPath] = count - 1
+		fmt.Printf("文件夹监听订阅数减少: %s (订阅数: %d)\n", absPath, count-1)
+		return nil
+	}
+
+	// 订阅数为 0，移除监听
+	// 只有在监听器运行时才调用 Remove
+	if isRunning {
+		if err := fsw.watcher.Remove(absPath); err != nil {
+			// 忽略 "can't remove non-existent watcher" 等错误
+			fmt.Printf("警告: 移除监听时出现错误 (忽略): %v\n", err)
+		}
+	}
+
+	delete(fsw.watchedPaths, absPath)
+	fmt.Printf("停止监听文件夹: %s\n", absPath)
+
+	return nil
+}
+
+// GetWatchedPaths 获取所有监听的路径
+func (fsw *FileSystemWatcher) GetWatchedPaths() []string {
+	fsw.pathsMutex.RLock()
+	defer fsw.pathsMutex.RUnlock()
+
+	paths := make([]string, 0, len(fsw.watchedPaths))
+	for path := range fsw.watchedPaths {
+		paths = append(paths, path)
+	}
+	return paths
+}
+
+// Stop 停止监听
+func (fsw *FileSystemWatcher) Stop() {
+	fsw.runMutex.Lock()
+	defer fsw.runMutex.Unlock()
+
+	if !fsw.running {
+		return
+	}
+
+	close(fsw.stopChan)
+	fsw.watcher.Close()
+	fsw.running = false
+}
+
+// Close 关闭监听器（Stop 的别名，用于资源清理）
+func (fsw *FileSystemWatcher) Close() {
+	fsw.Stop()
+}
+
+// ConvertEventToJSON 将事件转换为 JSON
+func (fsw *FileSystemWatcher) ConvertEventToJSON(event *FileSystemEvent) (string, error) {
+	data, err := json.Marshal(event)
+	if err != nil {
+		return "", fmt.Errorf("序列化事件失败: %w", err)
+	}
+	return string(data), nil
+}

src/rear/logs/app.log

@@ -98,3 +98,47 @@
 {"level":"info","time":"2025-08-24 16:51:39.825","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
 {"level":"info","time":"2025-08-24 17:07:03.334","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
 {"level":"info","time":"2025-08-24 17:07:41.666","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-03 13:06:31.665","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-03 14:38:12.220","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-03 15:59:26.845","caller":"config/config.go:90","msg":"加载配置文件成功","path":"D:\\go-argus\\src\\rear\\config.yaml"}
+{"level":"info","time":"2025-10-03 16:00:27.401","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-03 21:02:43.205","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-03 21:04:10.560","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 01:03:32.147","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 01:09:53.745","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 01:13:26.017","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 01:16:15.369","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 14:12:15.103","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 14:15:28.369","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 14:22:07.764","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:14:41.237","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:31:16.877","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:40:40.160","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:44:08.389","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:45:28.669","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:45:31.306","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:50:45.365","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:56:56.778","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 15:59:24.747","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 16:06:25.640","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 16:09:47.498","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 16:53:06.182","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:00:47.780","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:05:13.143","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:07:11.303","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:07:22.188","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:07:41.185","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:11:14.689","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:17:48.430","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 17:47:47.774","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-04 19:06:01.446","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 00:09:09.343","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 00:22:31.207","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 13:24:24.940","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 13:55:43.860","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 14:00:20.639","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 15:04:25.884","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 15:09:54.391","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 15:09:57.198","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 15:09:58.986","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}
+{"level":"info","time":"2025-10-05 15:29:14.093","caller":"config/config.go:90","msg":"加载配置文件成功","path":"config.yaml"}

src/rear/main.go

@@ -230,16 +230,17 @@ func startHttp(con *container.DbContainer, imgContain *container.TaskContainer)
 		pprof.Register(r)
 	}
 
-	// 设置路由
-	router.SetupRoutes(r, con, imgContain)
+	// 设置路由，并获取清理函数
+	cleanup := router.SetupRoutes(r, con, imgContain)
 
-	// 创建HTTP服务器
+	// 创建主 HTTP 服务器（所有请求，WriteTimeout = 0）
+	// 注意：WriteTimeout = 0 适用于本地使用，生产环境建议使用反向代理分离
 	srv := &http.Server{
 		Addr:         ":" + config.CONFIG.Port,
 		Handler:      r,
-		ReadTimeout:  config.CONFIG.ReadTimeout,
-		WriteTimeout: config.CONFIG.WriteTimeout,
-		IdleTimeout:  config.CONFIG.IdleTimeout,
+		ReadTimeout:  config.CONFIG.ReadTimeout,  // 30秒
+		WriteTimeout: 0,                          // 0 = 无限期（本地使用）
+		IdleTimeout:  config.CONFIG.IdleTimeout,  // 10分钟
 	}
 
 	// 优雅关闭
@@ -248,7 +249,7 @@ func startHttp(con *container.DbContainer, imgContain *container.TaskContainer)
 
 	// 启动服务器
 	go func() {
-		logger.Infof("Server starting on port 127.0.0.1:%s", config.CONFIG.Port)
+		logger.Infof("Server starting on 127.0.0.1:%s (WriteTimeout=0 for SSE support)", config.CONFIG.Port)
 		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			logger.Fatalf("Failed to start server: %v", err)
 			// 发送信号给主goroutine，让它知道启动失败
@@ -260,14 +261,20 @@ func startHttp(con *container.DbContainer, imgContain *container.TaskContainer)
 	<-quit
 	logger.Info("Shutting down server...")
 
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	// 先清理资源（关闭 SSE 连接和文件监听器）
+	logger.Info("Cleaning up resources...")
+	cleanup()
+	logger.Info("Resources cleaned up")
+
+	// 然后关闭 HTTP 服务器
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
 	if err := srv.Shutdown(ctx); err != nil {
-		logger.Fatalf("Server forced to shutdown: %v", err)
+		logger.Errorf("Server forced to shutdown: %v", err)
+	} else {
+		logger.Info("Server exited gracefully")
 	}
-
-	logger.Info("Server exited")
 }
 
 // initializeToolDependencies 初始化工具依赖（仅在开发阶段）

src/rear/pkg/sse/manager.go

@@ -2,6 +2,7 @@ package sse
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -23,6 +24,10 @@ type Client struct {
 	UserAgent   string
 	RemoteAddr  string
 	ConnectedAt time.Time
+
+	// 订阅管理
+	subscriptions     map[string]bool // key: 订阅的路径, value: true
+	subscriptionMutex sync.RWMutex    // 保护 subscriptions
 }
 
 type Event struct {
@@ -192,18 +197,48 @@ func (m *Manager) removeClient(clientID string) {
 
 func (m *Manager) broadcastToAllClients(event *Event) {
 	m.clientsMutex.RLock()
-	clients := make([]*Client, 0, len(m.clients))
+	clientCount := len(m.clients)
+
+	// 如果没有客户端，直接返回（避免向空列表广播）
+	if clientCount == 0 {
+		m.clientsMutex.RUnlock()
+		fmt.Printf("跳过广播: 当前无客户端连接\n")
+		return
+	}
+
+	clients := make([]*Client, 0, clientCount)
 	for _, client := range m.clients {
 		clients = append(clients, client)
 	}
 	m.clientsMutex.RUnlock()
 
+	// 并发发送，避免单个客户端阻塞
 	for _, client := range clients {
-		select {
-		case client.EventChan <- event:
-		case <-time.After(100 * time.Millisecond):
-			go m.UnregisterClient(client.ID)
-		}
+		go func(c *Client) {
+			// 双重检查：发送前再次确认客户端未关闭
+			select {
+			case <-c.CloseChan:
+				// 客户端已关闭，跳过
+				fmt.Printf("跳过向已关闭客户端 %s 发送事件\n", c.ID)
+				return
+			default:
+				// 继续发送
+			}
+
+			// 尝试发送事件，带超时和关闭检测
+			select {
+			case c.EventChan <- event:
+				// 发送成功
+			case <-time.After(100 * time.Millisecond):
+				// 发送超时，断开客户端
+				fmt.Printf("向客户端 %s 发送事件超时，断开连接\n", c.ID)
+				m.UnregisterClient(c.ID)
+			case <-c.CloseChan:
+				// 发送过程中客户端关闭
+				fmt.Printf("发送过程中客户端 %s 关闭\n", c.ID)
+				return
+			}
+		}(client)
 	}
 }
 
@@ -213,9 +248,14 @@ func (m *Manager) pingAllClients() {
 		Data:  fmt.Sprintf(`{"time":"%s"}`, time.Now().Format(time.RFC3339)),
 	}
 
+	// 记录心跳发送
+	fmt.Printf("发送心跳事件 (客户端数: %d)\n", len(m.clients))
+
 	select {
 	case m.broadcastChan <- pingEvent:
+		fmt.Printf("心跳事件已加入广播队列\n")
 	case <-time.After(1 * time.Second):
+		fmt.Printf("心跳事件发送超时\n")
 	}
 }
 
@@ -228,37 +268,110 @@ func (m *Manager) RegisterClient(c *gin.Context) {
 	c.Header("Cache-Control", "no-cache")
 	c.Header("Connection", "keep-alive")
 	c.Header("Access-Control-Allow-Origin", "*")
+	c.Header("X-Accel-Buffering", "no") // 禁用 Nginx 缓冲
 
 	clientID := uuid.New().String()
 	client := &Client{
-		ID:          clientID,
-		Writer:      c.Writer,
-		Request:     c.Request,
-		Context:     c,
-		EventChan:   make(chan *Event, 50),
-		CloseChan:   make(chan struct{}),
-		LastPing:    time.Now(),
-		UserAgent:   c.GetHeader("User-Agent"),
-		RemoteAddr:  c.ClientIP(),
-		ConnectedAt: time.Now(),
+		ID:            clientID,
+		Writer:        c.Writer,
+		Request:       c.Request,
+		Context:       c,
+		EventChan:     make(chan *Event, 50),
+		CloseChan:     make(chan struct{}),
+		LastPing:      time.Now(),
+		UserAgent:     c.GetHeader("User-Agent"),
+		RemoteAddr:    c.ClientIP(),
+		ConnectedAt:   time.Now(),
+		subscriptions: make(map[string]bool), // 初始化订阅映射
 	}
 
 	m.registerChan <- client
 
+	// 立即发送一个连接成功事件
+	connectEvent := &Event{
+		Event: "connected",
+		Data:  fmt.Sprintf(`{"client_id":"%s","time":"%s"}`, clientID, time.Now().Format(time.RFC3339)),
+	}
+	select {
+	case client.EventChan <- connectEvent:
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	// 使用 context 来确保资源清理
+	streamCtx, streamCancel := context.WithCancel(c.Request.Context())
+	defer streamCancel()
+
+	// 使用独立的 goroutine 发送 keepalive
+	go func() {
+		ticker := time.NewTicker(5 * time.Second) // 5秒发送一次 keepalive
+		defer ticker.Stop()
+		defer fmt.Printf("Keepalive goroutine exited for client %s\n", clientID)
+
+		for {
+			select {
+			case <-ticker.C:
+				// 双重检查：发送前确认客户端未关闭
+				select {
+				case <-client.CloseChan:
+					return
+				case <-streamCtx.Done():
+					return
+				default:
+				}
+
+				keepAliveEvent := &Event{
+					Event: "keepalive",
+					Data:  fmt.Sprintf(`{"time":"%s"}`, time.Now().Format(time.RFC3339)),
+				}
+
+				// 非阻塞发送，避免阻塞 keepalive 协程
+				select {
+				case client.EventChan <- keepAliveEvent:
+					// 成功发送，不打印日志（减少噪音）
+				case <-client.CloseChan:
+					return
+				case <-streamCtx.Done():
+					return
+				default:
+					// EventChan 已满，跳过本次 keepalive
+					fmt.Printf("Keepalive skipped for client %s (channel full)\n", clientID)
+				}
+
+			case <-client.CloseChan:
+				return
+			case <-streamCtx.Done():
+				return
+			case <-m.ctx.Done():
+				return
+			}
+		}
+	}()
+
 	c.Stream(func(w io.Writer) bool {
 		select {
 		case <-client.CloseChan:
+			fmt.Printf("Client %s CloseChan triggered in Stream\n", clientID)
+			streamCancel()
 			return false
 		case event := <-client.EventChan:
-			m.writeEvent(w, event)
+			// 写入事件，如果失败则断开连接
+			err := m.writeEvent(w, event)
+			if err != nil {
+				fmt.Printf("Failed to write SSE event to client %s: %v\n", clientID, err)
+				streamCancel() // 取消 context，通知 keepalive 协程退出
+				go m.UnregisterClient(clientID)
+				return false
+			}
 			client.LastPing = time.Now()
 			return true
-		case <-c.Request.Context().Done():
-			go m.UnregisterClient(clientID)
+		case <-streamCtx.Done():
+			fmt.Printf("Client %s context cancelled in Stream\n", clientID)
 			return false
 		}
 	})
 
+	// Stream 结束后确保客户端被注销
+	fmt.Printf("Stream ended for client %s, unregistering\n", clientID)
 	go m.UnregisterClient(clientID)
 }
 
@@ -281,21 +394,56 @@ func (m *Manager) isAllowedOrigin(origin string) bool {
 	return false
 }
 
-func (m *Manager) writeEvent(w io.Writer, event *Event) {
-	if event.ID != "" {
-		fmt.Fprintf(w, "id: %s\n", event.ID)
-	}
-	if event.Event != "" {
-		fmt.Fprintf(w, "event: %s\n", event.Event)
-	}
-	if event.Retry > 0 {
-		fmt.Fprintf(w, "retry: %d\n", event.Retry)
-	}
-	fmt.Fprintf(w, "data: %s\n\n", event.Data)
+func (m *Manager) writeEvent(w io.Writer, event *Event) error {
+	// 防御性编程：捕获写入错误
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Printf("SSE writeEvent panic recovered: %v, event: %+v\n", r, event)
+		}
+	}()
 
+	if event == nil {
+		return nil
+	}
+
+	// 检查 Writer 是否可用
+	if w == nil {
+		return fmt.Errorf("writer is nil")
+	}
+
+	var err error
+
+	// 一次性构建完整的 SSE 消息，减少多次 Fprintf 调用
+	var message string
+
+	if event.ID != "" {
+		message += fmt.Sprintf("id: %s\n", event.ID)
+	}
+
+	if event.Event != "" {
+		message += fmt.Sprintf("event: %s\n", event.Event)
+	}
+
+	if event.Retry > 0 {
+		message += fmt.Sprintf("retry: %d\n", event.Retry)
+	}
+
+	message += fmt.Sprintf("data: %s\n\n", event.Data)
+
+	// 一次性写入完整消息
+	n, err := fmt.Fprint(w, message)
+	if err != nil {
+		return fmt.Errorf("failed to write SSE event: %w (bytes written: %d)", err, n)
+	}
+
+	// 立即刷新缓冲区
 	if flusher, ok := w.(http.Flusher); ok {
 		flusher.Flush()
+	} else {
+		return fmt.Errorf("writer does not support flushing")
 	}
+
+	return nil
 }
 
 func (m *Manager) UnregisterClient(clientID string) {
@@ -318,15 +466,40 @@ func (m *Manager) Broadcast(message string) {
 }
 
 func (m *Manager) BroadcastEvent(eventType, message string) {
+	// 验证数据有效性
+	if message == "" {
+		fmt.Printf("Warning: 尝试广播空消息，事件类型: %s\n", eventType)
+		return
+	}
+
+	// 验证JSON格式
+	if eventType == "filesystem-change" {
+		var testData interface{}
+		if err := json.Unmarshal([]byte(message), &testData); err != nil {
+			fmt.Printf("Error: 广播事件数据JSON格式无效: %v, 数据: %s\n", err, message)
+			return
+		}
+	}
+
 	event := &Event{
 		ID:    uuid.New().String(),
 		Event: eventType,
 		Data:  message,
 	}
 
+	// 安全打印日志
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Printf("BroadcastEvent日志打印panic: %v\n", r)
+		}
+	}()
+
 	select {
 	case m.broadcastChan <- event:
+		// 使用安全的字符串格式化
+		fmt.Printf("SSE事件已发送: type=%q, data_length=%d\n", eventType, len(message))
 	case <-time.After(1 * time.Second):
+		fmt.Printf("Warning: SSE事件发送超时: type=%q\n", eventType)
 	}
 }
 
@@ -402,3 +575,143 @@ func (m *Manager) Close() {
 	close(m.unregisterChan)
 	close(m.eventChan)
 }
+
+// ============ 订阅管理方法 ============
+
+// Subscribe 客户端订阅指定路径的文件系统变化
+func (m *Manager) Subscribe(clientID, path string) error {
+	m.clientsMutex.RLock()
+	client, exists := m.clients[clientID]
+	m.clientsMutex.RUnlock()
+
+	if !exists {
+		return fmt.Errorf("客户端不存在: %s", clientID)
+	}
+
+	client.subscriptionMutex.Lock()
+	client.subscriptions[path] = true
+	client.subscriptionMutex.Unlock()
+
+	fmt.Printf("客户端 %s 订阅了路径: %s\n", clientID, path)
+	return nil
+}
+
+// Unsubscribe 客户端取消订阅指定路径
+func (m *Manager) Unsubscribe(clientID, path string) error {
+	m.clientsMutex.RLock()
+	client, exists := m.clients[clientID]
+	m.clientsMutex.RUnlock()
+
+	if !exists {
+		return fmt.Errorf("客户端不存在: %s", clientID)
+	}
+
+	client.subscriptionMutex.Lock()
+	delete(client.subscriptions, path)
+	client.subscriptionMutex.Unlock()
+
+	fmt.Printf("客户端 %s 取消订阅路径: %s\n", clientID, path)
+	return nil
+}
+
+// GetSubscriptions 获取客户端的所有订阅
+func (m *Manager) GetSubscriptions(clientID string) ([]string, error) {
+	m.clientsMutex.RLock()
+	client, exists := m.clients[clientID]
+	m.clientsMutex.RUnlock()
+
+	if !exists {
+		return nil, fmt.Errorf("客户端不存在: %s", clientID)
+	}
+
+	client.subscriptionMutex.RLock()
+	defer client.subscriptionMutex.RUnlock()
+
+	subs := make([]string, 0, len(client.subscriptions))
+	for path := range client.subscriptions {
+		subs = append(subs, path)
+	}
+
+	return subs, nil
+}
+
+// IsSubscribed 检查客户端是否订阅了指定路径
+func (m *Manager) IsSubscribed(clientID, path string) bool {
+	m.clientsMutex.RLock()
+	client, exists := m.clients[clientID]
+	m.clientsMutex.RUnlock()
+
+	if !exists {
+		return false
+	}
+
+	client.subscriptionMutex.RLock()
+	defer client.subscriptionMutex.RUnlock()
+
+	return client.subscriptions[path]
+}
+
+// BroadcastToSubscribers 向订阅了指定路径的客户端广播事件
+func (m *Manager) BroadcastToSubscribers(path, eventType, message string) {
+	// 验证数据有效性
+	if message == "" {
+		fmt.Printf("Warning: 尝试广播空消息，事件类型: %s, 路径: %s\n", eventType, path)
+		return
+	}
+
+	// 验证JSON格式
+	if eventType == "filesystem-change" {
+		var testData interface{}
+		if err := json.Unmarshal([]byte(message), &testData); err != nil {
+			fmt.Printf("Error: 广播事件数据JSON格式无效: %v, 数据: %s\n", err, message)
+			return
+		}
+	}
+
+	event := &Event{
+		ID:    uuid.New().String(),
+		Event: eventType,
+		Data:  message,
+	}
+
+	m.clientsMutex.RLock()
+	subscribedClients := make([]*Client, 0)
+
+	for _, client := range m.clients {
+		client.subscriptionMutex.RLock()
+		isSubscribed := client.subscriptions[path]
+		client.subscriptionMutex.RUnlock()
+
+		if isSubscribed {
+			subscribedClients = append(subscribedClients, client)
+		}
+	}
+	m.clientsMutex.RUnlock()
+
+	if len(subscribedClients) == 0 {
+		fmt.Printf("跳过广播: 没有客户端订阅路径 %s\n", path)
+		return
+	}
+
+	fmt.Printf("向 %d 个订阅了路径 %s 的客户端广播事件\n", len(subscribedClients), path)
+
+	// 并发发送
+	for _, client := range subscribedClients {
+		go func(c *Client) {
+			select {
+			case <-c.CloseChan:
+				return
+			default:
+			}
+
+			select {
+			case c.EventChan <- event:
+				// 发送成功
+			case <-time.After(100 * time.Millisecond):
+				fmt.Printf("向客户端 %s 发送事件超时\n", c.ID)
+			case <-c.CloseChan:
+				return
+			}
+		}(client)
+	}
+}

src/rear/pkg/sse/middleware.go

@@ -0,0 +1,56 @@
+package sse
+
+import (
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/gin-gonic/gin"
+)
+
+// NoWriteDeadlineMiddleware SSE 专用中间件：移除写超时限制
+// 使用方法: router.GET("/sse/connect", sse.NoWriteDeadlineMiddleware(), handler)
+func NoWriteDeadlineMiddleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		// 尝试获取底层的 TCP 连接
+		if conn, ok := getUnderlyingConn(c.Writer); ok {
+			// 设置一个非常远的未来时间作为 deadline（等同于无限期）
+			// 这会覆盖 http.Server 的 WriteTimeout
+			farFuture := time.Now().Add(876000 * time.Hour) // 100 年后
+			conn.SetWriteDeadline(farFuture)
+		}
+
+		c.Next()
+	}
+}
+
+// getUnderlyingConn 获取底层的 net.Conn
+func getUnderlyingConn(w gin.ResponseWriter) (net.Conn, bool) {
+	// Gin 的 ResponseWriter 层次结构：
+	// gin.ResponseWriter -> http.ResponseWriter -> http.response -> net.Conn
+
+	// 方法1: 尝试通过 Hijacker 接口获取连接
+	if hijacker, ok := w.(http.Hijacker); ok {
+		// 注意：这里不能真的调用 Hijack()，因为那会接管连接
+		// 我们只是检查接口是否支持
+
+		// 方法2: 通过反射或类型断言访问内部字段
+		// 但这依赖于 Go 的内部实现，不够稳定
+
+		// 更稳定的方法：依赖 http.response 的 Hijack 实现
+		// 实际上我们无法直接获取 net.Conn 而不 Hijack
+
+		_ = hijacker
+	}
+
+	// 由于无法直接访问 net.Conn，我们返回 false
+	// 真正的解决方案是在创建 http.Server 时就设置 WriteTimeout = 0
+	return nil, false
+}
+
+// SetNoWriteDeadline 尝试为当前连接移除写超时（已弃用，无法实现）
+// 这个方法无法在不 Hijack 连接的情况下修改 deadline
+func SetNoWriteDeadline(c *gin.Context) {
+	// 无法实现：无法在不 Hijack 的情况下访问 net.Conn
+	// 最佳方案是在 http.Server 创建时设置 WriteTimeout = 0
+}

src/rear/tests/sse_subscription.http

@@ -0,0 +1,56 @@
+### SSE 订阅管理 API 测试
+
+### 1. 连接 SSE（浏览器中手动测试）
+### 在浏览器控制台运行：
+### const eventSource = new EventSource('http://127.0.0.1:9484/api/v1/sse/connect')
+### eventSource.onmessage = (e) => console.log(e.data)
+
+### 2. 获取所有客户端
+GET http://127.0.0.1:9484/api/v1/sse/clients
+Content-Type: application/json
+
+###
+
+### 3. 订阅路径（请先从上一步获取 client_id）
+POST http://127.0.0.1:9484/api/v1/sse/subscribe
+Content-Type: application/json
+
+{
+  "client_id": "your-client-id-here",
+  "path": "D:\\EdgeDownload"
+}
+
+###
+
+### 4. 获取客户端订阅列表
+GET http://127.0.0.1:9484/api/v1/sse/subscriptions/your-client-id-here
+Content-Type: application/json
+
+###
+
+### 5. 取消订阅
+POST http://127.0.0.1:9484/api/v1/sse/unsubscribe
+Content-Type: application/json
+
+{
+  "client_id": "your-client-id-here",
+  "path": "D:\\EdgeDownload"
+}
+
+###
+
+### 6. 测试文件系统变化监听
+POST http://127.0.0.1:9484/api/v1/filesystem/watch
+Content-Type: application/json
+
+{
+  "path": "D:\\EdgeDownload"
+}
+
+###
+
+### 7. 停止文件系统监听
+DELETE http://127.0.0.1:9484/api/v1/filesystem/watch?path=D:\EdgeDownload
+Content-Type: application/json
+
+###