From 6d335e393d4579614140a4a580cdd110f549abd0 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Thu, 26 Mar 2026 21:17:49 +0800 Subject: [PATCH] feat(websocket): integrate WebSocket support and refactor SSE handling - Added WebSocket support by introducing a new WSHandler and related infrastructure. - Replaced SSE implementations with WebSocket equivalents across message and QR code handlers. - Updated service and repository layers to utilize WebSocket for real-time messaging. - Removed obsolete SSE hub and related code, streamlining the application for WebSocket usage. - Enhanced router to include WebSocket endpoints for real-time communication. --- ARCHITECTURE.md | 353 ++++++++++++++++++++ cmd/server/wire.go | 2 + cmd/server/wire_gen.go | 7 +- go.mod | 1 + go.sum | 2 + internal/handler/message_handler.go | 89 +----- internal/handler/qrcode_handler.go | 12 +- internal/handler/ws_handler.go | 389 +++++++++++++++++++++++ internal/model/call.go | 113 +++++++ internal/pkg/sse/hub.go | 152 --------- internal/pkg/ws/hub.go | 318 ++++++++++++++++++ internal/router/router.go | 9 +- internal/service/chat_service.go | 38 +-- internal/service/group_service.go | 12 +- internal/service/push_service.go | 28 +- internal/service/qrcode_login_service.go | 20 +- internal/wire/handler.go | 17 +- internal/wire/infrastructure.go | 12 +- internal/wire/service.go | 18 +- 19 files changed, 1277 insertions(+), 315 deletions(-) create mode 100644 ARCHITECTURE.md create mode 100644 internal/handler/ws_handler.go create mode 100644 internal/model/call.go delete mode 100644 internal/pkg/sse/hub.go create mode 100644 internal/pkg/ws/hub.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 0000000..dafd28e --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,353 @@ +# 后端架构设计规范 + +## 1. 分层架构原则 + +### 层级定义 +``` +Handler (控制器层) → Service (业务逻辑层) → Repository (数据访问层) → Infrastructure (基础设施层) +``` + +### 层级职责 + +| 层级 | 职责 | 依赖关系 | +|------|------|----------| +| Handler | 接收HTTP请求、参数校验、调用Service、返回响应 | 仅依赖Service | +| Service | 实现业务逻辑、事务管理、组合多个Repository | 仅依赖Repository接口 | +| Repository | 数据CRUD操作、数据库事务管理 | 仅依赖Model和GORM | +| Infrastructure | 缓存、消息队列、RPC等外部服务 | 被Service调用 | + +### 关键约束 +- **Handler禁止直接依赖Repository或直接操作数据库** +- **Service禁止直接依赖`gorm.DB`，必须通过Repository间接操作** +- **所有外部依赖通过接口抽象** + +--- + +## 2. Repository 层规范 + +### 2.1 接口 + 私有实现模式 + +所有Repository必须采用接口定义 + 私有实现结构体的模式： + +```go +// ✅ 正确：接口 + 私有实现 +type UserRepository interface { + Create(user *model.User) error + GetByID(id string) (*model.User, error) +} + +type userRepository struct { + db *gorm.DB +} + +func NewUserRepository(db *gorm.DB) UserRepository { + return &userRepository{db: db} +} + +func (r *userRepository) Create(user *model.User) error { + return r.db.Create(user).Error +} +``` + +```go +// ❌ 错误：直接暴露结构体指针 +type UserRepository struct { + db *gorm.DB +} + +func NewUserRepository(db *gorm.DB) *UserRepository { + return &UserRepository{db: db} +} +``` + +### 2.2 Repository方法设计原则 + +- 方法签名中**使用接口类型而非指针**：`func (r *userRepository) GetByID(...) (*model.User, error)` +- **避免在Repository层直接返回指针切片**，使用值切片更安全 +- **事务操作在Repository内部封装**，Service不应直接调用`db.Transaction()` + +### 2.3 新增Repository流程 + +1. 在`repository/`目录下创建`xxx_repo.go`文件 +2. 定义`XxxRepository`接口，包含所有数据访问方法 +3. 创建私有实现结构体`xxxRepository` +4. 创建构造函数`NewXxxRepository(db *gorm.DB) XxxRepository` +5. 在`wire/repository.go`的`RepositorySet`中注册 + +--- + +## 3. Service 层规范 + +### 3.1 接口 + 私有实现模式 + +```go +// ✅ 正确 +type UserService interface { + CreateUser(ctx context.Context, req *dto.CreateUserRequest) (*model.User, error) + GetUserByID(ctx context.Context, id string) (*model.User, error) +} + +type userService struct { + userRepo repository.UserRepository + // 其他Repo都使用接口类型 +} + +func NewUserService(userRepo repository.UserRepository) UserService { + return &userService{userRepo: userRepo} +} +``` + +### 3.2 Service依赖管理 + +- **所有Repository依赖必须使用接口类型**，禁止使用`*repository.XxxRepository`指针 +- **禁止在Service中直接持有`gorm.DB`**，如有数据库操作必须下沉到Repository +- **如需事务支持**，在Repository中封装事务方法，Service调用封装好的方法 + +### 3.3 Service间依赖 + +- Service之间通过接口依赖，避免循环依赖 +- 如Service A需要调用Service B的方法，确保该方法在接口中定义 + +--- + +## 4. Handler 层规范 + +### 4.1 Handler只依赖Service + +```go +// ✅ 正确 +type UserHandler struct { + userService service.UserService // 仅依赖Service接口 +} + +func NewUserHandler(userService service.UserService) *UserHandler { + return &UserHandler{userService: userService} +} +``` + +```go +// ❌ 错误：直接依赖Repository或Cache +type UserHandler struct { + userRepo *repository.UserRepository // 禁止 + cache cache.Cache // 禁止 +} +``` + +### 4.2 参数和类型分离 + +- **Query参数和Filter类型**应定义在`dto/`包中，而非`repository/`包 +- Handler负责将HTTP参数转换为DTO后传递给Service + +--- + +## 5. DTO/Filter 类型规范 + +### 5.1 放置位置 + +- **所有查询参数、过滤条件、分页参数**定义在`internal/dto/`目录下 +- **禁止在`repository/`包中定义供Handler使用的Filter类型** + +### 5.2 文件组织 + +``` +internal/dto/ +├── filter.go # 通用过滤条件（LogFilter, LoginFilter等） +├── user_converter.go # 用户相关转换 +└── ... +``` + +### 5.3 Filter类型定义示例 + +```go +// internal/dto/filter.go + +type LogFilter struct { + UserID string + Operation string + TargetType string + StartTime string + EndTime string +} + +type PaginationParams struct { + Page int + PageSize int +} +``` + +--- + +## 6. Wire 依赖注入规范 + +### 6.1 Provider函数设计 + +```go +// ✅ 正确：Provider返回接口类型 +func ProvideUserService( + userRepo repository.UserRepository, // 接口类型 + cacheBackend cache.Cache, +) service.UserService { + return service.NewUserService(userRepo, cacheBackend) +} + +// ❌ 错误：返回指针类型 +func ProvideUserService(userRepo *repository.UserRepository) *service.UserService +``` + +### 6.2 NewXxxService构造函数签名 + +- **构造函数参数使用接口类型** +- **返回值使用接口类型** + +```go +// ✅ 正确 +func NewUserService( + userRepo repository.UserRepository, // 接口 + cache cache.Cache, +) UserService { // 返回接口 + return &userService{userRepo: userRepo} +} +``` + +### 6.3 禁止的直接注入 + +- **禁止在Handler的Provider中直接注入Repository和Cache**，应仅注入Service +- **禁止在Service的Provider中直接注入`gorm.DB`** + +--- + +## 7. 事务管理规范 + +### 7.1 事务边界在Repository层 + +```go +// ✅ 正确：事务封装在Repository内部 +func (r *userRepository) CreateUserWithProfile(user *model.User, profile *model.Profile) error { + return r.db.Transaction(func(tx *gorm.DB) error { + if err := tx.Create(user).Error; err != nil { + return err + } + profile.UserID = user.ID + return tx.Create(profile).Error + }) +} +``` + +### 7.2 Service调用事务方法 + +```go +// ✅ Service通过调用Repository的封装方法使用事务 +func (s *userService) CreateUser(ctx context.Context, req *dto.CreateUserRequest) error { + user := &model.User{Name: req.Name} + profile := &model.Profile{Bio: req.Bio} + return s.userRepo.CreateUserWithProfile(user, profile) +} +``` + +--- + +## 8. 常见错误示例 + +### 错误1：Handler直接依赖Repository + +```go +// ❌ 错误 +type PostHandler struct { + postRepo *repository.PostRepository +} + +// ✅ 正确 +type PostHandler struct { + postService service.PostService +} +``` + +### 错误2：Service直接使用gorm.DB + +```go +// ❌ 错误 +type postService struct { + db *gorm.DB +} + +func (s *postService) CreatePost(post *model.Post) error { + return s.db.Create(post).Error +} + +// ✅ 正确：使用Repository +func (s *postService) CreatePost(post *model.Post) error { + return s.postRepo.Create(post) +} +``` + +### 错误3：Repository返回指针类型 + +```go +// ❌ 错误 +func NewUserRepository(db *gorm.DB) *UserRepository { + return &UserRepository{db: db} +} + +// ✅ 正确 +func NewUserRepository(db *gorm.DB) UserRepository { + return &userRepository{db: db} +} +``` + +### 错误4：Filter类型放在repository包 + +```go +// ❌ 错误：repository包不应该被handler直接引用 +package repository +type LogFilter struct { + UserID string +} + +// ✅ 正确：放在dto包 +package dto +type LogFilter struct { + UserID string +} +``` + +--- + +## 9. 新功能开发 Checklist + +开发新功能时，确保满足以下规范： + +- [ ] **Repository定义** + - [ ] 接口 + 私有实现结构体 + - [ ] 构造函数返回接口类型 + - [ ] 方法参数使用值类型而非指针 + - [ ] 在`wire/repository.go`中注册 + +- [ ] **Service定义** + - [ ] 接口 + 私有实现结构体 + - [ ] 所有Repository依赖使用接口类型 + - [ ] 不直接持有`gorm.DB` + - [ ] 事务操作下沉到Repository + +- [ ] **Handler定义** + - [ ] 仅依赖Service接口 + - [ ] 不直接依赖Repository或Cache + - [ ] Query/Filter参数使用dto包类型 + +- [ ] **Wire Provider** + - [ ] 参数使用接口类型 + - [ ] 返回值使用接口类型 + - [ ] 不直接注入`gorm.DB`到Service + +--- + +## 10. 相关文件位置参考 + +| 类别 | 目录 | +|------|------| +| Handler | `internal/handler/` | +| Service | `internal/service/` | +| Repository | `internal/repository/` | +| DTO/Filter | `internal/dto/` | +| Model | `internal/model/` | +| Wire | `internal/wire/` | +| Cache | `internal/cache/` | diff --git a/cmd/server/wire.go b/cmd/server/wire.go index d2054bd..dc90369 100644 --- a/cmd/server/wire.go +++ b/cmd/server/wire.go @@ -50,6 +50,7 @@ func ProvideRouter( logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, + wsHandler *handler.WSHandler, ) *router.Router { return router.New( userHandler, @@ -78,5 +79,6 @@ func ProvideRouter( logService, activityService, casbinService, + wsHandler, ) } diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index 12a96f2..9dcba62 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -31,7 +31,7 @@ func InitializeApp() (*App, error) { pushRecordRepository := repository.NewPushRecordRepository(db) deviceTokenRepository := repository.NewDeviceTokenRepository(db) messageRepository := repository.NewMessageRepository(db) - hub := wire.ProvideSSEHub() + hub := wire.ProvideWSHub() pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, hub) postRepository := repository.NewPostRepository(db) client := wire.ProvideRedisClient(config) @@ -115,11 +115,12 @@ func InitializeApp() (*App, error) { adminLogHandler := handler.NewAdminLogHandler(operationLogService, loginLogService, dataChangeLogService) qrCodeLoginService := wire.ProvideQRCodeLoginService(client, hub, jwtService, userService, userActivityService, logService) qrCodeHandler := handler.NewQRCodeHandler(qrCodeLoginService) + wsHandler := handler.NewWSHandler(hub, chatService, groupService, jwtService) materialSubjectRepository := repository.NewMaterialSubjectRepository(db) materialFileRepository := repository.NewMaterialFileRepository(db) materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository) materialHandler := handler.NewMaterialHandler(materialService) - router := ProvideRouter(userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, logService, userActivityService, casbinService) + router := ProvideRouter(userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, logService, userActivityService, casbinService, wsHandler) hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, cache) app := NewApp(config, db, router, pushService, hotRankWorker, server, logger) return app, nil @@ -155,6 +156,7 @@ func ProvideRouter( logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, + wsHandler *handler.WSHandler, ) *router.Router { return router.New( userHandler, @@ -183,5 +185,6 @@ func ProvideRouter( logService, activityService, casbinService, + wsHandler, ) } diff --git a/go.mod b/go.mod index 1a28737..1aeb383 100644 --- a/go.mod +++ b/go.mod @@ -51,6 +51,7 @@ require ( github.com/goccy/go-yaml v1.19.2 // indirect github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 // indirect github.com/golang-sql/sqlexp v0.1.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/pgx/v5 v5.8.0 // indirect diff --git a/go.sum b/go.sum index dc0f158..60a4add 100644 --- a/go.sum +++ b/go.sum @@ -124,6 +124,8 @@ github.com/google/wire v0.7.0 h1:JxUKI6+CVBgCO2WToKy/nQk0sS+amI9z9EjVmdaocj4= github.com/google/wire v0.7.0/go.mod h1:n6YbUQD9cPKTnHXEBN2DXlOp/mVADhVErcMFb0v3J18= github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 31b142d..cb8a388 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -2,10 +2,7 @@ package handler import ( "context" - "fmt" - "net/http" "strconv" - "time" "github.com/gin-gonic/gin" @@ -13,7 +10,7 @@ import ( "carrot_bbs/internal/model" "carrot_bbs/internal/pkg/cursor" "carrot_bbs/internal/pkg/response" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/service" ) @@ -23,95 +20,17 @@ type MessageHandler struct { messageService *service.MessageService userService service.UserService groupService service.GroupService - sseHub *sse.Hub + wsHub *ws.Hub } // NewMessageHandler 创建消息处理器 -func NewMessageHandler(chatService service.ChatService, messageService *service.MessageService, userService service.UserService, groupService service.GroupService, sseHub *sse.Hub) *MessageHandler { +func NewMessageHandler(chatService service.ChatService, messageService *service.MessageService, userService service.UserService, groupService service.GroupService, wsHub *ws.Hub) *MessageHandler { return &MessageHandler{ chatService: chatService, messageService: messageService, userService: userService, groupService: groupService, - sseHub: sseHub, - } -} - -// HandleSSE 实时消息订阅（SSE） -// GET /api/v1/realtime/sse -func (h *MessageHandler) HandleSSE(c *gin.Context) { - userID := c.GetString("user_id") - if userID == "" { - response.Unauthorized(c, "") - return - } - if h.sseHub == nil { - response.InternalServerError(c, "sse hub not available") - return - } - - lastID := sse.ParseEventID(c.GetHeader("Last-Event-ID")) - if lastID == 0 { - lastID = sse.ParseEventID(c.Query("last_event_id")) - } - ch, cancel, replay := h.sseHub.Subscribe(userID, lastID) - defer cancel() - - w := c.Writer - flusher, ok := w.(http.Flusher) - if !ok { - response.InternalServerError(c, "streaming unsupported") - return - } - - w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache, no-transform") - w.Header().Set("Connection", "keep-alive") - w.Header().Set("X-Accel-Buffering", "no") - c.Status(http.StatusOK) - flusher.Flush() - - writeEvent := func(ev sse.Event) bool { - data, err := sse.EncodeData(ev) - if err != nil { - return false - } - if _, err := fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", ev.ID, ev.Event, data); err != nil { - return false - } - flusher.Flush() - return true - } - - for _, ev := range replay { - if !writeEvent(ev) { - return - } - } - - // Use a shorter heartbeat interval to survive common 50-60s proxy idle timeout. - heartbeat := time.NewTicker(10 * time.Second) - defer heartbeat.Stop() - - for { - select { - case <-c.Request.Context().Done(): - return - case ev, ok := <-ch: - if !ok || !writeEvent(ev) { - return - } - case <-heartbeat.C: - // Comment keepalive line for SSE intermediaries/proxies. - if _, err := fmt.Fprint(w, ": keepalive\n\n"); err != nil { - return - } - flusher.Flush() - if _, err := fmt.Fprint(w, "event: heartbeat\ndata: {}\n\n"); err != nil { - return - } - flusher.Flush() - } + wsHub: wsHub, } } diff --git a/internal/handler/qrcode_handler.go b/internal/handler/qrcode_handler.go index d0ad844..642bb50 100644 --- a/internal/handler/qrcode_handler.go +++ b/internal/handler/qrcode_handler.go @@ -8,7 +8,7 @@ import ( "github.com/gin-gonic/gin" "carrot_bbs/internal/pkg/response" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/service" ) @@ -44,16 +44,16 @@ func (h *QRCodeHandler) GetQRCode(c *gin.Context) { }) } -// SSEEvents SSE事件流 +// WSEvents WebSocket事件流 // GET /api/v1/auth/qrcode/events?session_id=xxx -func (h *QRCodeHandler) SSEEvents(c *gin.Context) { +func (h *QRCodeHandler) WSEvents(c *gin.Context) { sessionID := c.Query("session_id") if sessionID == "" { response.BadRequest(c, "session_id is required") return } - ch, cancel, replay := h.qrcodeService.GetSSEHub().Subscribe(sessionID, 0) + ch, cancel, replay := h.qrcodeService.GetWSHub().Subscribe(sessionID, 0) defer cancel() w := c.Writer @@ -70,8 +70,8 @@ func (h *QRCodeHandler) SSEEvents(c *gin.Context) { c.Status(http.StatusOK) flusher.Flush() - writeEvent := func(ev sse.Event) bool { - data, err := sse.EncodeData(ev) + writeEvent := func(ev ws.Event) bool { + data, err := ws.EncodeData(ev) if err != nil { return false } diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go new file mode 100644 index 0000000..61adf5b --- /dev/null +++ b/internal/handler/ws_handler.go @@ -0,0 +1,389 @@ +package handler + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "sync/atomic" + "time" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "go.uber.org/zap" + + "carrot_bbs/internal/dto" + "carrot_bbs/internal/model" + "carrot_bbs/internal/pkg/response" + "carrot_bbs/internal/pkg/ws" + "carrot_bbs/internal/service" +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 4096, + CheckOrigin: func(r *http.Request) bool { + return true // 允许所有来源，实际生产环境应该限制 + }, + HandshakeTimeout: 10 * time.Second, +} + +// WSHandler WebSocket处理器 +type WSHandler struct { + wsHub *ws.Hub + chatService service.ChatService + groupService service.GroupService + jwtService *service.JWTService + clientSeq uint64 +} + +// NewWSHandler 创建WebSocket处理器 +func NewWSHandler( + wsHub *ws.Hub, + chatService service.ChatService, + groupService service.GroupService, + jwtService *service.JWTService, +) *WSHandler { + return &WSHandler{ + wsHub: wsHub, + chatService: chatService, + groupService: groupService, + jwtService: jwtService, + } +} + +// HandleWebSocket WebSocket连接处理 +// GET /api/v1/realtime/ws?token=xxx +func (h *WSHandler) HandleWebSocket(c *gin.Context) { + // 1. 认证 - 支持Query参数和Header两种方式 + token := c.Query("token") + if token == "" { + authHeader := c.GetHeader("Authorization") + if authHeader != "" { + parts := strings.SplitN(authHeader, " ", 2) + if len(parts) == 2 && parts[0] == "Bearer" { + token = parts[1] + } + } + } + + if token == "" { + response.Unauthorized(c, "token is required") + return + } + + claims, err := h.jwtService.ParseToken(token) + if err != nil { + response.Unauthorized(c, "invalid token") + return + } + + userID := claims.UserID + + // 2. 升级HTTP连接为WebSocket + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + zap.L().Error("Failed to upgrade WebSocket connection", + zap.String("user_id", userID), + zap.Error(err), + ) + return + } + + // 3. 创建客户端 + clientID := atomic.AddUint64(&h.clientSeq, 1) + client := &ws.Client{ + ID: clientID, + UserID: userID, + Send: make(chan []byte, defaultUserBufferSize), + Quit: make(chan struct{}), + } + + // 4. 注册客户端 + replayEvents := h.wsHub.Register(client) + defer h.wsHub.Unregister(client) + + // 5. 发送历史回放消息 + go func() { + for _, ev := range replayEvents { + msg := ws.ResponseMessage{ + EventID: ev.ID, + Type: ev.Type, + TS: ev.TS, + Payload: ev.Payload, + } + data, _ := json.Marshal(msg) + select { + case <-client.Quit: + return + case client.Send <- data: + } + } + }() + + // 6. 启动读写goroutine + go h.writePump(conn, client) + h.readPump(conn, client) +} + +// readPump 读取客户端消息 +func (h *WSHandler) readPump(conn *websocket.Conn, client *ws.Client) { + defer func() { + conn.Close() + close(client.Quit) + }() + + conn.SetReadLimit(64 * 1024) // 64KB + conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + conn.SetPongHandler(func(string) error { + conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + return nil + }) + + for { + _, message, err := conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + zap.L().Debug("WebSocket read error", + zap.String("user_id", client.UserID), + zap.Error(err), + ) + } + break + } + + // 解析消息 + var msg ws.Message + if err := json.Unmarshal(message, &msg); err != nil { + h.wsHub.SendError(client, "parse_error", "invalid message format") + continue + } + + // 处理消息 + h.handleMessage(client, &msg) + } +} + +// writePump 向客户端发送消息 +func (h *WSHandler) writePump(conn *websocket.Conn, client *ws.Client) { + ticker := time.NewTicker(30 * time.Second) + defer func() { + ticker.Stop() + conn.Close() + }() + + for { + select { + case <-client.Quit: + conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + case message, ok := <-client.Send: + conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if !ok { + conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + w, err := conn.NextWriter(websocket.TextMessage) + if err != nil { + return + } + w.Write(message) + + // 批量发送缓冲区中的消息 + n := len(client.Send) + for i := 0; i < n; i++ { + w.Write([]byte{'\n'}) + w.Write(<-client.Send) + } + + if err := w.Close(); err != nil { + return + } + case <-ticker.C: + conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} + +// handleMessage 处理客户端消息 +func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) { + ctx := context.Background() + + switch msg.Type { + case "ping": + h.handlePing(client) + case "chat": + h.handleChat(ctx, client, msg.Payload) + case "typing": + h.handleTyping(ctx, client, msg.Payload) + case "read": + h.handleRead(ctx, client, msg.Payload) + case "recall": + h.handleRecall(ctx, client, msg.Payload) + default: + h.wsHub.SendError(client, "unknown_type", fmt.Sprintf("unknown message type: %s", msg.Type)) + } +} + +// handlePing 处理ping消息 +func (h *WSHandler) handlePing(client *ws.Client) { + pong := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "pong", + TS: time.Now().UnixMilli(), + } + data, _ := json.Marshal(pong) + select { + case <-client.Quit: + case client.Send <- data: + } +} + +// handleChat 处理聊天消息 +func (h *WSHandler) handleChat(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + ConversationID string `json:"conversation_id"` + DetailType string `json:"detail_type"` + Segments model.MessageSegments `json:"segments"` + ReplyToID *string `json:"reply_to_id,omitempty"` + } + + if err := json.Unmarshal(payload, &req); err != nil { + h.wsHub.SendError(client, "parse_error", "invalid chat message format") + return + } + + if req.ConversationID == "" { + h.wsHub.SendError(client, "invalid_params", "conversation_id is required") + return + } + + if len(req.Segments) == 0 { + h.wsHub.SendError(client, "invalid_params", "segments is required") + return + } + + // 发送消息 + message, err := h.chatService.SendMessage(ctx, client.UserID, req.ConversationID, req.Segments, req.ReplyToID) + if err != nil { + h.wsHub.SendError(client, "send_failed", err.Error()) + return + } + + // 发送成功响应 + detailType := "private" + if conv, _ := h.chatService.GetConversationByID(ctx, req.ConversationID, client.UserID); conv != nil && conv.Type == model.ConversationTypeGroup { + detailType = "group" + } + + resp := dto.WSEventResponse{ + ID: message.ID, + Time: message.CreatedAt.UnixMilli(), + Type: "message", + DetailType: detailType, + Seq: strconv.FormatInt(message.Seq, 10), + Segments: req.Segments, + SenderID: client.UserID, + } + + msg := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "message_sent", + TS: time.Now().UnixMilli(), + Payload: resp, + } + data, _ := json.Marshal(msg) + select { + case <-client.Quit: + case client.Send <- data: + } +} + +// handleTyping 处理输入状态 +func (h *WSHandler) handleTyping(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + ConversationID string `json:"conversation_id"` + IsTyping bool `json:"is_typing"` + } + + if err := json.Unmarshal(payload, &req); err != nil { + return + } + + if req.ConversationID == "" { + return + } + + // 输入状态通常不需要持久化，直接通过ChatService的SendTyping处理 + // 这里简化处理，直接推送 + h.chatService.SendTyping(ctx, client.UserID, req.ConversationID) +} + +// handleRead 处理已读标记 +func (h *WSHandler) handleRead(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + ConversationID string `json:"conversation_id"` + Seq int64 `json:"seq"` + } + + if err := json.Unmarshal(payload, &req); err != nil { + return + } + + if req.ConversationID == "" || req.Seq <= 0 { + return + } + + err := h.chatService.MarkAsRead(ctx, req.ConversationID, client.UserID, req.Seq) + if err != nil { + zap.L().Debug("Failed to mark as read via WebSocket", + zap.String("user_id", client.UserID), + zap.String("conversation_id", req.ConversationID), + zap.Error(err), + ) + } +} + +// handleRecall 处理消息撤回 +func (h *WSHandler) handleRecall(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + MessageID string `json:"message_id"` + } + + if err := json.Unmarshal(payload, &req); err != nil { + h.wsHub.SendError(client, "parse_error", "invalid recall message format") + return + } + + if req.MessageID == "" { + h.wsHub.SendError(client, "invalid_params", "message_id is required") + return + } + + err := h.chatService.RecallMessage(ctx, req.MessageID, client.UserID) + if err != nil { + h.wsHub.SendError(client, "recall_failed", err.Error()) + return + } + + // 发送撤回成功响应 + resp := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "message_recalled", + TS: time.Now().UnixMilli(), + Payload: map[string]string{"message_id": req.MessageID}, + } + data, _ := json.Marshal(resp) + select { + case <-client.Quit: + case client.Send <- data: + } +} + +const defaultUserBufferSize = 128 diff --git a/internal/model/call.go b/internal/model/call.go new file mode 100644 index 0000000..3e79f8b --- /dev/null +++ b/internal/model/call.go @@ -0,0 +1,113 @@ +package model + +import ( + "strconv" + "time" + + "gorm.io/gorm" + + "carrot_bbs/internal/pkg/utils" +) + +// CallType 通话类型 +type CallType string + +const ( + CallTypePrivate CallType = "private" // 私聊通话 + CallTypeGroup CallType = "group" // 群聊通话 +) + +// CallStatus 通话状态 +type CallStatus string + +const ( + CallStatusCalling CallStatus = "calling" // 呼叫中 + CallStatusConnected CallStatus = "connected" // 已接通 + CallStatusEnded CallStatus = "ended" // 已结束 + CallStatusRejected CallStatus = "rejected" // 已拒绝 + CallStatusMissed CallStatus = "missed" // 未接听 + CallStatusCancelled CallStatus = "cancelled" // 已取消 +) + +// ParticipantStatus 参与者状态 +type ParticipantStatus string + +const ( + ParticipantStatusInvited ParticipantStatus = "invited" // 已邀请 + ParticipantStatusJoined ParticipantStatus = "joined" // 已加入 + ParticipantStatusLeft ParticipantStatus = "left" // 已离开 + ParticipantStatusRejected ParticipantStatus = "rejected" // 已拒绝 +) + +// CallSession 通话会话 +type CallSession struct { + ID string `gorm:"primaryKey;size:20" json:"id"` // 雪花算法ID + ConversationID string `gorm:"not null;size:20;index" json:"conversation_id"` // 会话ID + GroupID *string `gorm:"index" json:"group_id,omitempty"` // 群组ID（群聊时使用） + CallerID string `gorm:"column:caller_id;type:varchar(50);index;not null" json:"caller_id"` // 发起者ID + CallType CallType `gorm:"type:varchar(20);not null" json:"call_type"` // 通话类型 + Status CallStatus `gorm:"type:varchar(20);default:'calling'" json:"status"` // 通话状态 + StartedAt *time.Time `json:"started_at,omitempty"` // 接通时间 + EndedAt *time.Time `json:"ended_at,omitempty"` // 结束时间 + Duration int64 `gorm:"default:0" json:"duration"` // 通话时长（秒） + CreatedAt time.Time `json:"created_at" gorm:"autoCreateTime"` + UpdatedAt time.Time `json:"updated_at" gorm:"autoUpdateTime"` + DeletedAt gorm.DeletedAt `json:"-" gorm:"index"` + + // 关联 + Participants []CallParticipant `gorm:"foreignKey:CallID" json:"participants,omitempty"` +} + +// BeforeCreate 创建前生成雪花算法ID +func (c *CallSession) BeforeCreate(tx *gorm.DB) error { + if c.ID == "" { + id, err := utils.GetSnowflake().GenerateID() + if err != nil { + return err + } + c.ID = strconv.FormatInt(id, 10) + } + return nil +} + +func (CallSession) TableName() string { + return "call_sessions" +} + +// CallParticipant 通话参与者 +type CallParticipant struct { + ID uint `gorm:"primaryKey" json:"id"` + CallID string `gorm:"not null;size:20;index" json:"call_id"` // 通话ID + UserID string `gorm:"column:user_id;type:varchar(50);index;not null" json:"user_id"` // 用户ID + Status ParticipantStatus `gorm:"type:varchar(20);default:'invited'" json:"status"` // 参与状态 + JoinedAt *time.Time `json:"joined_at,omitempty"` // 加入时间 + LeftAt *time.Time `json:"left_at,omitempty"` // 离开时间 + CreatedAt time.Time `json:"created_at" gorm:"autoCreateTime"` + UpdatedAt time.Time `json:"updated_at" gorm:"autoUpdateTime"` +} + +func (CallParticipant) TableName() string { + return "call_participants" +} + +// IsPrivate 是否私聊通话 +func (c *CallSession) IsPrivate() bool { + return c.CallType == CallTypePrivate +} + +// IsGroup 是否群聊通话 +func (c *CallSession) IsGroup() bool { + return c.CallType == CallTypeGroup +} + +// IsActive 是否活跃通话 +func (c *CallSession) IsActive() bool { + return c.Status == CallStatusCalling || c.Status == CallStatusConnected +} + +// CalculateDuration 计算通话时长 +func (c *CallSession) CalculateDuration() { + if c.StartedAt != nil && c.EndedAt != nil { + c.Duration = int64(c.EndedAt.Sub(*c.StartedAt).Seconds()) + } +} diff --git a/internal/pkg/sse/hub.go b/internal/pkg/sse/hub.go deleted file mode 100644 index 57585f7..0000000 --- a/internal/pkg/sse/hub.go +++ /dev/null @@ -1,152 +0,0 @@ -package sse - -import ( - "encoding/json" - "strconv" - "sync" - "sync/atomic" - "time" -) - -const ( - defaultUserBufferSize = 128 - maxReplayEvents = 200 -) - -type Event struct { - ID uint64 `json:"event_id"` - Event string `json:"event"` - TS int64 `json:"ts"` - Payload interface{} `json:"payload"` -} - -type subscriber struct { - id uint64 - ch chan Event - quit chan struct{} -} - -type Hub struct { - seq uint64 - - mu sync.RWMutex - subscribers map[string]map[uint64]*subscriber - history map[string][]Event -} - -func NewHub() *Hub { - return &Hub{ - subscribers: make(map[string]map[uint64]*subscriber), - history: make(map[string][]Event), - } -} - -func (h *Hub) NextID() uint64 { - return atomic.AddUint64(&h.seq, 1) -} - -func ParseEventID(raw string) uint64 { - if raw == "" { - return 0 - } - id, err := strconv.ParseUint(raw, 10, 64) - if err != nil { - return 0 - } - return id -} - -func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Event) { - subID := h.NextID() - sub := &subscriber{ - id: subID, - ch: make(chan Event, defaultUserBufferSize), - quit: make(chan struct{}), - } - - h.mu.Lock() - if _, ok := h.subscribers[userID]; !ok { - h.subscribers[userID] = make(map[uint64]*subscriber) - } - h.subscribers[userID][subID] = sub - - replay := make([]Event, 0) - for _, e := range h.history[userID] { - if e.ID > afterID { - replay = append(replay, e) - } - } - h.mu.Unlock() - - cancel := func() { - h.mu.Lock() - defer h.mu.Unlock() - if userSubs, ok := h.subscribers[userID]; ok { - if s, exists := userSubs[subID]; exists { - close(s.quit) - delete(userSubs, subID) - close(s.ch) - } - if len(userSubs) == 0 { - delete(h.subscribers, userID) - } - } - } - - return sub.ch, cancel, replay -} - -func (h *Hub) HasSubscribers(userID string) bool { - h.mu.RLock() - defer h.mu.RUnlock() - return len(h.subscribers[userID]) > 0 -} - -func (h *Hub) PublishToUser(userID string, eventName string, payload interface{}) Event { - ev := Event{ - ID: h.NextID(), - Event: eventName, - TS: time.Now().UnixMilli(), - Payload: payload, - } - h.publish(userID, ev) - return ev -} - -func (h *Hub) PublishToUsers(userIDs []string, eventName string, payload interface{}) { - for _, uid := range userIDs { - h.PublishToUser(uid, eventName, payload) - } -} - -func (h *Hub) publish(userID string, ev Event) { - h.mu.Lock() - history := append(h.history[userID], ev) - if len(history) > maxReplayEvents { - history = history[len(history)-maxReplayEvents:] - } - h.history[userID] = history - - targets := make([]*subscriber, 0, len(h.subscribers[userID])) - for _, s := range h.subscribers[userID] { - targets = append(targets, s) - } - h.mu.Unlock() - - for _, s := range targets { - select { - case <-s.quit: - case s.ch <- ev: - default: - // 慢消费者丢弃单条消息，客户端可通过 Last-Event-ID + HTTP 同步补偿 - } - } -} - -func EncodeData(ev Event) (string, error) { - body, err := json.Marshal(ev.Payload) - if err != nil { - return "", err - } - return string(body), nil -} diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go new file mode 100644 index 0000000..3627f9d --- /dev/null +++ b/internal/pkg/ws/hub.go @@ -0,0 +1,318 @@ +package ws + +import ( + "encoding/json" + "sync" + "sync/atomic" + "time" + + "go.uber.org/zap" +) + +const ( + defaultUserBufferSize = 128 + maxReplayEvents = 200 +) + +// Event 表示一个WebSocket事件 +type Event struct { + ID uint64 `json:"event_id"` + Type string `json:"type"` + Event string `json:"event"` // 事件名称（兼容多种协议） + TS int64 `json:"ts"` + Payload interface{} `json:"payload"` +} + +// Client 表示一个WebSocket客户端连接 +type Client struct { + ID uint64 + UserID string + Send chan []byte + Quit chan struct{} +} + +// Message 表示WebSocket消息 +type Message struct { + Type string `json:"type"` + Payload json.RawMessage `json:"payload"` +} + +// ResponseMessage 表示发送给客户端的消息 +type ResponseMessage struct { + EventID uint64 `json:"event_id"` + Type string `json:"type"` + TS int64 `json:"ts"` + Payload interface{} `json:"payload,omitempty"` +} + +// ErrorMessage 表示错误消息 +type ErrorMessage struct { + Type string `json:"type"` + Payload struct { + Code string `json:"code"` + Message string `json:"message"` + } `json:"payload"` +} + +// subscriber 表示一个事件订阅者 +type subscriber struct { + id uint64 + ch chan Event + quit chan struct{} +} + +// Hub WebSocket连接管理器 +type Hub struct { + seq uint64 + + mu sync.RWMutex + clients map[string]map[uint64]*Client // userID -> clientID -> Client + history map[string][]Event // userID -> []Event (用于断线重连历史回放) + subscribers map[string]map[uint64]*subscriber // userID -> subscriberID -> subscriber +} + +// NewHub 创建WebSocket Hub +func NewHub() *Hub { + return &Hub{ + clients: make(map[string]map[uint64]*Client), + history: make(map[string][]Event), + subscribers: make(map[string]map[uint64]*subscriber), + } +} + +// NextID 获取下一个事件ID +func (h *Hub) NextID() uint64 { + return atomic.AddUint64(&h.seq, 1) +} + +// Register 注册客户端连接 +func (h *Hub) Register(client *Client) []Event { + h.mu.Lock() + defer h.mu.Unlock() + + if _, ok := h.clients[client.UserID]; !ok { + h.clients[client.UserID] = make(map[uint64]*Client) + } + h.clients[client.UserID][client.ID] = client + + // 获取历史事件用于回放 + replay := make([]Event, 0) + for _, e := range h.history[client.UserID] { + replay = append(replay, e) + } + + zap.L().Debug("WebSocket client registered", + zap.String("user_id", client.UserID), + zap.Uint64("client_id", client.ID), + zap.Int("total_clients", len(h.clients[client.UserID])), + ) + + return replay +} + +// Unregister 注销客户端连接 +func (h *Hub) Unregister(client *Client) { + h.mu.Lock() + defer h.mu.Unlock() + + if userClients, ok := h.clients[client.UserID]; ok { + if _, exists := userClients[client.ID]; exists { + delete(userClients, client.ID) + close(client.Quit) + + if len(userClients) == 0 { + delete(h.clients, client.UserID) + } + + zap.L().Debug("WebSocket client unregistered", + zap.String("user_id", client.UserID), + zap.Uint64("client_id", client.ID), + ) + } + } +} + +// HasClients 检查用户是否有活跃连接 +func (h *Hub) HasClients(userID string) bool { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.clients[userID]) > 0 +} + +// GetClientCount 获取用户的连接数 +func (h *Hub) GetClientCount(userID string) int { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.clients[userID]) +} + +// PublishToUser 向指定用户发送事件 +func (h *Hub) PublishToUser(userID string, eventType string, payload interface{}) Event { + ev := Event{ + ID: h.NextID(), + Type: eventType, + TS: time.Now().UnixMilli(), + Payload: payload, + } + h.publish(userID, ev) + return ev +} + +// PublishToUsers 向多个用户发送事件 +func (h *Hub) PublishToUsers(userIDs []string, eventType string, payload interface{}) { + for _, uid := range userIDs { + h.PublishToUser(uid, eventType, payload) + } +} + +// publish 内部发布方法 +func (h *Hub) publish(userID string, ev Event) { + h.mu.Lock() + // 存储历史事件（最多保留200条） + history := append(h.history[userID], ev) + if len(history) > maxReplayEvents { + history = history[len(history)-maxReplayEvents:] + } + h.history[userID] = history + + // 复制客户端列表避免锁竞争 + targets := make([]*Client, 0, len(h.clients[userID])) + for _, c := range h.clients[userID] { + targets = append(targets, c) + } + h.mu.Unlock() + + // 构建消息 + msg := ResponseMessage{ + EventID: ev.ID, + Type: ev.Type, + TS: ev.TS, + Payload: ev.Payload, + } + data, err := json.Marshal(msg) + if err != nil { + zap.L().Error("Failed to marshal WebSocket message", + zap.String("user_id", userID), + zap.String("event_type", ev.Type), + zap.Error(err), + ) + return + } + + // 非阻塞发送，慢消费者丢弃消息 + for _, c := range targets { + select { + case <-c.Quit: + // 客户端已断开 + case c.Send <- data: + // 消息已发送 + default: + // 发送缓冲区满，丢弃消息 + zap.L().Warn("WebSocket client buffer full, dropping message", + zap.String("user_id", userID), + zap.Uint64("client_id", c.ID), + ) + } + } +} + +// SendError 向客户端发送错误消息 +func (h *Hub) SendError(client *Client, code string, message string) { + errMsg := ErrorMessage{ + Type: "error", + } + errMsg.Payload.Code = code + errMsg.Payload.Message = message + + data, err := json.Marshal(errMsg) + if err != nil { + return + } + + select { + case <-client.Quit: + case client.Send <- data: + default: + } +} + +// Broadcast 广播消息给所有连接 +func (h *Hub) Broadcast(eventType string, payload interface{}) { + h.mu.RLock() + userIDs := make([]string, 0, len(h.clients)) + for uid := range h.clients { + userIDs = append(userIDs, uid) + } + h.mu.RUnlock() + + h.PublishToUsers(userIDs, eventType, payload) +} + +// GetOnlineUsers 获取在线用户列表 +func (h *Hub) GetOnlineUsers() []string { + h.mu.RLock() + defer h.mu.RUnlock() + + users := make([]string, 0, len(h.clients)) + for uid := range h.clients { + users = append(users, uid) + } + return users +} + +// GetOnlineCount 获取在线用户数 +func (h *Hub) GetOnlineCount() int { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.clients) +} + +// Subscribe 订阅事件 +func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Event) { + subID := h.NextID() + sub := &subscriber{ + id: subID, + ch: make(chan Event, defaultUserBufferSize), + quit: make(chan struct{}), + } + + h.mu.Lock() + if _, ok := h.subscribers[userID]; !ok { + h.subscribers[userID] = make(map[uint64]*subscriber) + } + h.subscribers[userID][subID] = sub + + replay := make([]Event, 0) + for _, e := range h.history[userID] { + if e.ID > afterID { + replay = append(replay, e) + } + } + h.mu.Unlock() + + cancel := func() { + h.mu.Lock() + defer h.mu.Unlock() + if userSubs, ok := h.subscribers[userID]; ok { + if s, exists := userSubs[subID]; exists { + close(s.quit) + delete(userSubs, subID) + close(s.ch) + } + if len(userSubs) == 0 { + delete(h.subscribers, userID) + } + } + } + + return sub.ch, cancel, replay +} + +// EncodeData 编码事件数据 +func EncodeData(ev Event) (string, error) { + body, err := json.Marshal(ev.Payload) + if err != nil { + return "", err + } + return string(body), nil +} \ No newline at end of file diff --git a/internal/router/router.go b/internal/router/router.go index d659d54..31de36a 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -34,6 +34,7 @@ type Router struct { adminLogHandler *handler.AdminLogHandler qrcodeHandler *handler.QRCodeHandler materialHandler *handler.MaterialHandler + wsHandler *handler.WSHandler logService *service.LogService jwtService *service.JWTService casbinService service.CasbinService @@ -67,6 +68,7 @@ func New( logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, + wsHandler *handler.WSHandler, ) *Router { // 设置JWT服务 userHandler.SetJWTService(jwtService) @@ -100,6 +102,7 @@ func New( logService: logService, jwtService: jwtService, casbinService: casbinService, + wsHandler: wsHandler, } r.setupRoutes() @@ -194,7 +197,7 @@ func (r *Router) setupRoutes() { qrcode := v1.Group("/auth/qrcode") { qrcode.GET("", r.qrcodeHandler.GetQRCode) - qrcode.GET("/events", r.qrcodeHandler.SSEEvents) + qrcode.GET("/events", r.qrcodeHandler.WSEvents) } // 二维码操作（需认证） @@ -327,9 +330,9 @@ func (r *Router) setupRoutes() { } realtime := v1.Group("/realtime") - realtime.Use(authMiddleware) { - realtime.GET("/sse", r.messageHandler.HandleSSE) + // WebSocket端点 - 在中间件之前处理认证（通过query参数） + realtime.GET("/ws", r.wsHandler.HandleWebSocket) } // 消息操作路由 diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index f2ca4bf..e72cb4b 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -10,7 +10,7 @@ import ( "carrot_bbs/internal/cache" "carrot_bbs/internal/dto" "carrot_bbs/internal/model" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" "go.uber.org/zap" @@ -59,7 +59,7 @@ type chatServiceImpl struct { repo repository.MessageRepository userRepo repository.UserRepository sensitive SensitiveService - sseHub *sse.Hub + wsHub *ws.Hub // 缓存相关字段 conversationCache *cache.ConversationCache @@ -71,7 +71,7 @@ func NewChatService( repo repository.MessageRepository, userRepo repository.UserRepository, sensitive SensitiveService, - sseHub *sse.Hub, + wsHub *ws.Hub, cacheBackend cache.Cache, uploadService *UploadService, ) ChatService { @@ -91,17 +91,17 @@ func NewChatService( repo: repo, userRepo: userRepo, sensitive: sensitive, - sseHub: sseHub, + wsHub: wsHub, conversationCache: conversationCache, uploadService: uploadService, } } -func (s *chatServiceImpl) publishSSEToUsers(userIDs []string, event string, payload interface{}) { - if s.sseHub == nil || len(userIDs) == 0 { +func (s *chatServiceImpl) publishToUsers(userIDs []string, event string, payload interface{}) { + if s.wsHub == nil || len(userIDs) == 0 { return } - s.sseHub.PublishToUsers(userIDs, event, payload) + s.wsHub.PublishToUsers(userIDs, event, payload) } // GetOrCreateConversation 获取或创建私聊会话 @@ -314,12 +314,12 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv } }() - // 获取会话中的参与者并发送 SSE + // 获取会话中的参与者并发送消息 participants, err := s.getParticipants(ctx, conversationID) if err == nil { targetIDs := make([]string, 0, len(participants)) for _, p := range participants { - // 私聊场景下，发送者已经从 HTTP 响应拿到消息，避免再通过 SSE 回推导致本端重复展示。 + // 私聊场景下，发送者已经从 HTTP 响应拿到消息，避免再通过 WebSocket 回推导致本端重复展示。 if conv.Type == model.ConversationTypePrivate && p.UserID == senderID { continue } @@ -329,7 +329,7 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv if conv.Type == model.ConversationTypeGroup { detailType = "group" } - s.publishSSEToUsers(targetIDs, "chat_message", map[string]interface{}{ + s.publishToUsers(targetIDs, "chat_message", map[string]interface{}{ "detail_type": detailType, "message": dto.ConvertMessageToResponse(message), }) @@ -342,7 +342,7 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv s.conversationCache.InvalidateUnreadCount(p.UserID, conversationID) } if totalUnread, uErr := s.repo.GetAllUnreadCount(p.UserID); uErr == nil { - s.publishSSEToUsers([]string{p.UserID}, "conversation_unread", map[string]interface{}{ + s.publishToUsers([]string{p.UserID}, "conversation_unread", map[string]interface{}{ "conversation_id": conversationID, "total_unread": totalUnread, }) @@ -490,7 +490,7 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, for _, p := range participants { targetIDs = append(targetIDs, p.UserID) } - s.publishSSEToUsers(targetIDs, "message_read", map[string]interface{}{ + s.publishToUsers(targetIDs, "message_read", map[string]interface{}{ "detail_type": detailType, "conversation_id": conversationID, "group_id": groupID, @@ -499,7 +499,7 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, }) } if totalUnread, uErr := s.repo.GetAllUnreadCount(userID); uErr == nil { - s.publishSSEToUsers([]string{userID}, "conversation_unread", map[string]interface{}{ + s.publishToUsers([]string{userID}, "conversation_unread", map[string]interface{}{ "conversation_id": conversationID, "total_unread": totalUnread, }) @@ -579,7 +579,7 @@ func (s *chatServiceImpl) RecallMessage(ctx context.Context, messageID string, u for _, p := range participants { targetIDs = append(targetIDs, p.UserID) } - s.publishSSEToUsers(targetIDs, "message_recall", map[string]interface{}{ + s.publishToUsers(targetIDs, "message_recall", map[string]interface{}{ "detail_type": detailType, "conversation_id": message.ConversationID, "group_id": groupID, @@ -627,7 +627,7 @@ func (s *chatServiceImpl) DeleteMessage(ctx context.Context, messageID string, u // SendTyping 发送正在输入状态 func (s *chatServiceImpl) SendTyping(ctx context.Context, senderID string, conversationID string) { - if s.sseHub == nil { + if s.wsHub == nil { return } @@ -651,8 +651,8 @@ func (s *chatServiceImpl) SendTyping(ctx context.Context, senderID string, conve if p.UserID == senderID { continue } - if s.sseHub != nil { - s.sseHub.PublishToUser(p.UserID, "typing", map[string]interface{}{ + if s.wsHub != nil { + s.wsHub.PublishToUser(p.UserID, "typing", map[string]interface{}{ "detail_type": detailType, "conversation_id": conversationID, "user_id": senderID, @@ -664,8 +664,8 @@ func (s *chatServiceImpl) SendTyping(ctx context.Context, senderID string, conve // IsUserOnline 检查用户是否在线 func (s *chatServiceImpl) IsUserOnline(userID string) bool { - if s.sseHub != nil { - return s.sseHub.HasSubscribers(userID) + if s.wsHub != nil { + return s.wsHub.HasClients(userID) } return false } diff --git a/internal/service/group_service.go b/internal/service/group_service.go index 19ed87b..63576ca 100644 --- a/internal/service/group_service.go +++ b/internal/service/group_service.go @@ -14,8 +14,8 @@ import ( apperrors "carrot_bbs/internal/errors" "carrot_bbs/internal/model" "carrot_bbs/internal/pkg/cursor" - "carrot_bbs/internal/pkg/sse" "carrot_bbs/internal/pkg/utils" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" "go.uber.org/zap" @@ -111,19 +111,19 @@ type groupService struct { messageRepo repository.MessageRepository requestRepo repository.GroupJoinRequestRepository notifyRepo repository.SystemNotificationRepository - sseHub *sse.Hub + wsHub *ws.Hub cache cache.Cache } // NewGroupService 创建群组服务 -func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.UserRepository, messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, sseHub *sse.Hub, cacheBackend cache.Cache) GroupService { +func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.UserRepository, messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, wsHub *ws.Hub, cacheBackend cache.Cache) GroupService { return &groupService{ groupRepo: groupRepo, userRepo: userRepo, messageRepo: messageRepo, requestRepo: requestRepo, notifyRepo: notifyRepo, - sseHub: sseHub, + wsHub: wsHub, cache: cacheBackend, } } @@ -152,9 +152,9 @@ func (s *groupService) publishGroupNotice(groupID string, notice groupNoticeMess ) return } - if s.sseHub != nil { + if s.wsHub != nil { for _, m := range members { - s.sseHub.PublishToUser(m.UserID, "group_notice", notice) + s.wsHub.PublishToUser(m.UserID, "group_notice", notice) } } } diff --git a/internal/service/push_service.go b/internal/service/push_service.go index 97f171a..d3d2de2 100644 --- a/internal/service/push_service.go +++ b/internal/service/push_service.go @@ -8,7 +8,7 @@ import ( "carrot_bbs/internal/dto" "carrot_bbs/internal/model" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" ) @@ -65,7 +65,7 @@ type pushServiceImpl struct { pushRepo repository.PushRecordRepository deviceRepo repository.DeviceTokenRepository messageRepo repository.MessageRepository - sseHub *sse.Hub + wsHub *ws.Hub // 推送队列 pushQueue chan *pushTask @@ -84,13 +84,13 @@ func NewPushService( pushRepo repository.PushRecordRepository, deviceRepo repository.DeviceTokenRepository, messageRepo repository.MessageRepository, - sseHub *sse.Hub, + wsHub *ws.Hub, ) PushService { return &pushServiceImpl{ pushRepo: pushRepo, deviceRepo: deviceRepo, messageRepo: messageRepo, - sseHub: sseHub, + wsHub: wsHub, pushQueue: make(chan *pushTask, PushQueueSize), stopChan: make(chan struct{}), } @@ -138,7 +138,7 @@ func (s *pushServiceImpl) PushToUser(ctx context.Context, userID string, message // pushViaWebSocket 通过WebSocket推送消息 // 返回true表示推送成功，false表示用户不在线 func (s *pushServiceImpl) pushViaWebSocket(ctx context.Context, userID string, message *model.Message) bool { - if s.sseHub == nil || !s.sseHub.HasSubscribers(userID) { + if s.wsHub == nil || !s.wsHub.HasClients(userID) { return false } @@ -174,7 +174,7 @@ func (s *pushServiceImpl) pushViaWebSocket(ctx context.Context, userID string, m } } } - s.sseHub.PublishToUser(userID, "system_notification", notification) + s.wsHub.PublishToUser(userID, "system_notification", notification) return true } @@ -199,7 +199,7 @@ func (s *pushServiceImpl) pushViaWebSocket(ctx context.Context, userID string, m SenderID: message.SenderID, } - s.sseHub.PublishToUser(userID, "chat_message", map[string]interface{}{ + s.wsHub.PublishToUser(userID, "chat_message", map[string]interface{}{ "detail_type": detailType, "message": event, }) @@ -444,7 +444,7 @@ func (s *pushServiceImpl) PushSystemMessage(ctx context.Context, userID string, // pushSystemViaWebSocket 通过WebSocket推送系统消息 func (s *pushServiceImpl) pushSystemViaWebSocket(ctx context.Context, userID string, msgType, title, content string, data map[string]interface{}) bool { - if s.sseHub == nil || !s.sseHub.HasSubscribers(userID) { + if s.wsHub == nil || !s.wsHub.HasClients(userID) { return false } @@ -455,7 +455,7 @@ func (s *pushServiceImpl) pushSystemViaWebSocket(ctx context.Context, userID str "data": data, "created_at": time.Now().UnixMilli(), } - s.sseHub.PublishToUser(userID, "system_notification", sysMsg) + s.wsHub.PublishToUser(userID, "system_notification", sysMsg) return true } @@ -472,11 +472,11 @@ func (s *pushServiceImpl) PushSystemNotification(ctx context.Context, userID str // pushSystemNotificationViaWebSocket 通过WebSocket推送系统通知 func (s *pushServiceImpl) pushSystemNotificationViaWebSocket(ctx context.Context, userID string, notification *model.SystemNotification) bool { - if s.sseHub == nil || !s.sseHub.HasSubscribers(userID) { + if s.wsHub == nil || !s.wsHub.HasClients(userID) { return false } - sseNotification := map[string]interface{}{ + wsNotification := map[string]interface{}{ "id": fmt.Sprintf("%d", notification.ID), "type": string(notification.Type), "title": notification.Title, @@ -487,7 +487,7 @@ func (s *pushServiceImpl) pushSystemNotificationViaWebSocket(ctx context.Context // 填充额外数据 if notification.ExtraData != nil { - extra := sseNotification["extra"].(map[string]interface{}) + extra := wsNotification["extra"].(map[string]interface{}) extra["actor_id_str"] = notification.ExtraData.ActorIDStr extra["actor_name"] = notification.ExtraData.ActorName extra["avatar_url"] = notification.ExtraData.AvatarURL @@ -498,7 +498,7 @@ func (s *pushServiceImpl) pushSystemNotificationViaWebSocket(ctx context.Context // 设置触发用户信息 if notification.ExtraData.ActorIDStr != "" { - sseNotification["trigger_user"] = map[string]interface{}{ + wsNotification["trigger_user"] = map[string]interface{}{ "id": notification.ExtraData.ActorIDStr, "username": notification.ExtraData.ActorName, "avatar": notification.ExtraData.AvatarURL, @@ -506,6 +506,6 @@ func (s *pushServiceImpl) pushSystemNotificationViaWebSocket(ctx context.Context } } - s.sseHub.PublishToUser(userID, "system_notification", sseNotification) + s.wsHub.PublishToUser(userID, "system_notification", wsNotification) return true } diff --git a/internal/service/qrcode_login_service.go b/internal/service/qrcode_login_service.go index 6dec665..ea896a5 100644 --- a/internal/service/qrcode_login_service.go +++ b/internal/service/qrcode_login_service.go @@ -9,7 +9,7 @@ import ( "carrot_bbs/internal/model" "carrot_bbs/internal/pkg/redis" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" ) // Lua 脚本：原子性地检查状态并更新为 scanned @@ -54,7 +54,7 @@ type QRCodeSession struct { // QRCodeLoginService 二维码登录服务 type QRCodeLoginService struct { redis *redis.Client - sseHub *sse.Hub + wsHub *ws.Hub jwtService *JWTService userService UserService activityService UserActivityService @@ -62,10 +62,10 @@ type QRCodeLoginService struct { } // NewQRCodeLoginService 创建二维码登录服务 -func NewQRCodeLoginService(redis *redis.Client, sseHub *sse.Hub, jwtService *JWTService, userService UserService, activityService UserActivityService, logService *LogService) *QRCodeLoginService { +func NewQRCodeLoginService(redis *redis.Client, wsHub *ws.Hub, jwtService *JWTService, userService UserService, activityService UserActivityService, logService *LogService) *QRCodeLoginService { return &QRCodeLoginService{ redis: redis, - sseHub: sseHub, + wsHub: wsHub, jwtService: jwtService, userService: userService, activityService: activityService, @@ -170,7 +170,7 @@ func (s *QRCodeLoginService) Scan(ctx context.Context, sessionID, userID string) "avatar": user.Avatar, }, } - s.sseHub.PublishToUser(sessionID, "scanned", payload) + s.wsHub.PublishToUser(sessionID, "scanned", payload) return nil } @@ -242,7 +242,7 @@ func (s *QRCodeLoginService) Confirm(ctx context.Context, sessionID, userID stri RefreshToken: refreshToken, User: user, } - s.sseHub.PublishToUser(sessionID, "confirmed", response) + s.wsHub.PublishToUser(sessionID, "confirmed", response) return response, nil } @@ -275,14 +275,14 @@ func (s *QRCodeLoginService) Cancel(ctx context.Context, sessionID, userID strin } // 推送取消事件 - s.sseHub.PublishToUser(sessionID, "cancelled", map[string]interface{}{}) + s.wsHub.PublishToUser(sessionID, "cancelled", map[string]interface{}{}) return nil } -// GetSSEHub 获取SSE Hub -func (s *QRCodeLoginService) GetSSEHub() *sse.Hub { - return s.sseHub +// GetWSHub 获取WS Hub +func (s *QRCodeLoginService) GetWSHub() *ws.Hub { + return s.wsHub } // GetUserService 获取用户服务 diff --git a/internal/wire/handler.go b/internal/wire/handler.go index deb519b..4a84583 100644 --- a/internal/wire/handler.go +++ b/internal/wire/handler.go @@ -2,7 +2,7 @@ package wire import ( "carrot_bbs/internal/handler" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/service" "github.com/google/wire" @@ -35,6 +35,7 @@ var HandlerSet = wire.NewSet( ProvideSystemMessageHandler, ProvideGroupHandler, ProvideScheduleHandler, + ProvideWSHandler, ) // ProvideUserHandler 提供用户处理器 @@ -91,9 +92,9 @@ func ProvideMessageHandler( messageService *service.MessageService, userService service.UserService, groupService service.GroupService, - sseHub *sse.Hub, + wsHub *ws.Hub, ) *handler.MessageHandler { - return handler.NewMessageHandler(chatService, messageService, userService, groupService, sseHub) + return handler.NewMessageHandler(chatService, messageService, userService, groupService, wsHub) } // ProvideSystemMessageHandler 提供系统消息处理器 @@ -118,3 +119,13 @@ func ProvideScheduleHandler( ) *handler.ScheduleHandler { return handler.NewScheduleHandler(scheduleService, scheduleSyncService) } + +// ProvideWSHandler 提供WebSocket处理器 +func ProvideWSHandler( + wsHub *ws.Hub, + chatService service.ChatService, + groupService service.GroupService, + jwtService *service.JWTService, +) *handler.WSHandler { + return handler.NewWSHandler(wsHub, chatService, groupService, jwtService) +} diff --git a/internal/wire/infrastructure.go b/internal/wire/infrastructure.go index 5fafb24..dfc3b3b 100644 --- a/internal/wire/infrastructure.go +++ b/internal/wire/infrastructure.go @@ -12,7 +12,7 @@ import ( "carrot_bbs/internal/pkg/openai" "carrot_bbs/internal/pkg/redis" "carrot_bbs/internal/pkg/s3" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" "carrot_bbs/internal/service" @@ -44,8 +44,8 @@ var InfrastructureSet = wire.NewSet( ProvideBuiltinHooks, ProvideModerationHooks, - // SSE Hub - ProvideSSEHub, + // WebSocket Hub + ProvideWSHub, // 外部服务客户端 ProvideOpenAIClient, @@ -156,9 +156,9 @@ func ProvideModerationHooks( return moderationHooks } -// ProvideSSEHub 提供 SSE Hub -func ProvideSSEHub() *sse.Hub { - return sse.NewHub() +// ProvideWSHub 提供 WebSocket Hub +func ProvideWSHub() *ws.Hub { + return ws.NewHub() } // ProvideOpenAIClient 提供 OpenAI 客户端 diff --git a/internal/wire/service.go b/internal/wire/service.go index d9cefdb..0b3ff2d 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -11,7 +11,7 @@ import ( "carrot_bbs/internal/pkg/openai" "carrot_bbs/internal/pkg/redis" "carrot_bbs/internal/pkg/s3" - "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" "carrot_bbs/internal/service" @@ -89,9 +89,9 @@ func ProvidePushService( pushRepo repository.PushRecordRepository, deviceTokenRepo repository.DeviceTokenRepository, messageRepo repository.MessageRepository, - sseHub *sse.Hub, + wsHub *ws.Hub, ) service.PushService { - return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, sseHub) + return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, wsHub) } // ProvideSystemMessageService 提供系统消息服务 @@ -183,11 +183,11 @@ func ProvideChannelService(channelRepo repository.ChannelRepository, cacheBacken func ProvideChatService( messageRepo repository.MessageRepository, userRepo repository.UserRepository, - sseHub *sse.Hub, + wsHub *ws.Hub, cacheBackend cache.Cache, uploadService *service.UploadService, ) service.ChatService { - return service.NewChatService(messageRepo, userRepo, nil, sseHub, cacheBackend, uploadService) + return service.NewChatService(messageRepo, userRepo, nil, wsHub, cacheBackend, uploadService) } // ProvideScheduleService 提供日程服务 @@ -213,10 +213,10 @@ func ProvideGroupService( messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, - sseHub *sse.Hub, + wsHub *ws.Hub, cacheBackend cache.Cache, ) service.GroupService { - return service.NewGroupService(groupRepo, userRepo, messageRepo, requestRepo, notifyRepo, sseHub, cacheBackend) + return service.NewGroupService(groupRepo, userRepo, messageRepo, requestRepo, notifyRepo, wsHub, cacheBackend) } // ProvideUploadService 提供上传服务 @@ -359,13 +359,13 @@ func ProvideHotRankWorker(cfg *config.Config, postRepo repository.PostRepository func ProvideQRCodeLoginService( redisClient *redis.Client, - sseHub *sse.Hub, + wsHub *ws.Hub, jwtService *service.JWTService, userService service.UserService, activityService service.UserActivityService, logService *service.LogService, ) *service.QRCodeLoginService { - return service.NewQRCodeLoginService(redisClient, sseHub, jwtService, userService, activityService, logService) + return service.NewQRCodeLoginService(redisClient, wsHub, jwtService, userService, activityService, logService) } // ProvideMaterialService 提供学习资料服务