diff --git a/backend/go-service/app/im/handler/event_handler.go b/backend/go-service/app/im/handler/event_handler.go new file mode 100644 index 0000000..69f94e5 --- /dev/null +++ b/backend/go-service/app/im/handler/event_handler.go @@ -0,0 +1,155 @@ +// Package handler 提供 IM 模块的 WebSocket 事件处理器 +// 通过 Hub.RegisterEvent 注册到事件路由表,处理 im.* 系列事件 +package handler + +import ( + "context" + "encoding/json" + + "github.com/echochat/backend/app/dto" + "github.com/echochat/backend/app/im/service" + "github.com/echochat/backend/pkg/logs" + "github.com/echochat/backend/pkg/ws" + "go.uber.org/zap" +) + +// EventHandler IM 模块的 WS 事件处理器 +type EventHandler struct { + imService *service.IMService + hub *ws.Hub +} + +// NewEventHandler 创建 EventHandler 实例并注册事件到 Hub 路由表 +func NewEventHandler(imService *service.IMService, hub *ws.Hub) *EventHandler { + h := &EventHandler{ + imService: imService, + hub: hub, + } + h.registerEvents() + return h +} + +// registerEvents 将 IM 事件处理函数注册到 Hub 路由表 +func (h *EventHandler) registerEvents() { + h.hub.RegisterEvent("im.message.send", h.handleSendMessage) + h.hub.RegisterEvent("im.message.recall", h.handleRecallMessage) + h.hub.RegisterEvent("im.conversation.read", h.handleMarkRead) + h.hub.RegisterEvent("im.typing", h.handleTyping) +} + +// handleSendMessage 处理发送消息事件 +func (h *EventHandler) handleSendMessage(client *ws.Client, msg *ws.Message) { + funcName := "handler.event_handler.handleSendMessage" + ctx := context.Background() + + var req dto.SendMessageRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + logs.Warn(ctx, funcName, "解析消息请求失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + h.sendACK(client, msg, -1, "请求参数格式错误", nil) + return + } + + msgDTO, err := h.imService.SendMessage(ctx, client.UserID, &req) + if err != nil { + if err == service.ErrDuplicateMsg && msgDTO != nil { + h.sendACK(client, msg, 0, "ok", msgDTO) + return + } + logs.Warn(ctx, funcName, "发送消息失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + h.sendACK(client, msg, -1, err.Error(), nil) + return + } + + h.sendACK(client, msg, 0, "ok", msgDTO) +} + +// handleRecallMessage 处理撤回消息事件 +func (h *EventHandler) handleRecallMessage(client *ws.Client, msg *ws.Message) { + funcName := "handler.event_handler.handleRecallMessage" + ctx := context.Background() + + var req dto.RecallMessageRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + logs.Warn(ctx, funcName, "解析撤回请求失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + h.sendACK(client, msg, -1, "请求参数格式错误", nil) + return + } + + if err := h.imService.RecallMessage(ctx, client.UserID, req.MessageID); err != nil { + logs.Warn(ctx, funcName, "撤回消息失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + h.sendACK(client, msg, -1, err.Error(), nil) + return + } + + h.sendACK(client, msg, 0, "ok", nil) +} + +// handleMarkRead 处理标记已读事件 +func (h *EventHandler) handleMarkRead(client *ws.Client, msg *ws.Message) { + funcName := "handler.event_handler.handleMarkRead" + ctx := context.Background() + + var req dto.MarkReadRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + logs.Warn(ctx, funcName, "解析已读请求失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + h.sendACK(client, msg, -1, "请求参数格式错误", nil) + return + } + + if err := h.imService.MarkRead(ctx, client.UserID, req.ConversationID); err != nil { + logs.Warn(ctx, funcName, "标记已读失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + h.sendACK(client, msg, -1, err.Error(), nil) + return + } + + h.sendACK(client, msg, 0, "ok", nil) +} + +// handleTyping 处理正在输入事件(转发给对方) +func (h *EventHandler) handleTyping(client *ws.Client, msg *ws.Message) { + funcName := "handler.event_handler.handleTyping" + ctx := context.Background() + + var req dto.TypingRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + logs.Warn(ctx, funcName, "解析输入状态请求失败", + zap.Int64("user_id", client.UserID), zap.Error(err)) + return + } + + peerID, err := h.imService.GetPeerUserID(ctx, req.ConversationID, client.UserID) + if err != nil { + logs.Warn(ctx, funcName, "查询对方用户 ID 失败", + zap.Int64("conversation_id", req.ConversationID), zap.Error(err)) + return + } + + push := ws.NewPushMessage("im.typing", map[string]interface{}{ + "conversation_id": req.ConversationID, + "user_id": client.UserID, + }) + data, err := ws.MarshalPush(push) + if err != nil { + return + } + + h.hub.SendToUser(peerID, data) +} + +// sendACK 发送 ACK 响应给客户端 +func (h *EventHandler) sendACK(client *ws.Client, msg *ws.Message, code int, message string, data interface{}) { + resp := ws.NewResponse(msg.Event, msg.Seq, code, message, data) + bytes, err := ws.MarshalResponse(resp) + if err != nil { + logs.Error(context.Background(), "handler.event_handler.sendACK", "序列化 ACK 失败", + zap.Error(err)) + return + } + client.Send(bytes) +} diff --git a/backend/go-service/app/im/handler/offline_pusher.go b/backend/go-service/app/im/handler/offline_pusher.go new file mode 100644 index 0000000..1cdb115 --- /dev/null +++ b/backend/go-service/app/im/handler/offline_pusher.go @@ -0,0 +1,84 @@ +package handler + +import ( + "context" + + "github.com/echochat/backend/app/im/dao" + "github.com/echochat/backend/app/im/service" + "github.com/echochat/backend/pkg/logs" + "github.com/echochat/backend/pkg/ws" + "go.uber.org/zap" +) + +// OfflinePusher 离线消息推送器 +// 在用户 WebSocket 重新连接后,主动推送未读会话摘要 +type OfflinePusher struct { + imService *service.IMService + convDAO *dao.ConversationDAO + pubsub *ws.PubSub +} + +// NewOfflinePusher 创建 OfflinePusher 实例 +func NewOfflinePusher(imService *service.IMService, convDAO *dao.ConversationDAO, pubsub *ws.PubSub) *OfflinePusher { + return &OfflinePusher{ + imService: imService, + convDAO: convDAO, + pubsub: pubsub, + } +} + +// PushOfflineMessages 向用户推送离线消息摘要 +// 包含所有有未读消息的会话信息和全局未读总数 +func (p *OfflinePusher) PushOfflineMessages(ctx context.Context, userID int64) { + funcName := "handler.offline_pusher.PushOfflineMessages" + logs.Info(ctx, funcName, "推送离线消息", zap.Int64("user_id", userID)) + + unreadConvs, err := p.convDAO.GetUnreadConversations(ctx, userID) + if err != nil { + logs.Error(ctx, funcName, "查询未读会话失败", zap.Error(err)) + return + } + + totalUnread, err := p.imService.GetTotalUnread(ctx, userID) + if err != nil { + logs.Error(ctx, funcName, "查询全局未读数失败", zap.Error(err)) + totalUnread = 0 + } + + if len(unreadConvs) == 0 && totalUnread == 0 { + return + } + + convList := make([]map[string]interface{}, 0, len(unreadConvs)) + for _, c := range unreadConvs { + item := map[string]interface{}{ + "conversation_id": c.ID, + "unread_count": c.UnreadCount, + "last_msg_content": c.LastMsgContent, + } + if c.LastMsgTime != nil { + item["last_msg_time"] = c.LastMsgTime.Format("2006-01-02 15:04:05") + } + convList = append(convList, item) + } + + push := ws.NewPushMessage("im.offline.sync", map[string]interface{}{ + "total_unread": totalUnread, + "conversations": convList, + }) + data, err := ws.MarshalPush(push) + if err != nil { + logs.Error(ctx, funcName, "序列化离线消息失败", zap.Error(err)) + return + } + + if err := p.pubsub.Publish(ctx, userID, data); err != nil { + logs.Error(ctx, funcName, "推送离线消息失败", + zap.Int64("user_id", userID), zap.Error(err)) + } + + logs.Info(ctx, funcName, "离线消息推送完成", + zap.Int64("user_id", userID), + zap.Int("unread_conv_count", len(unreadConvs)), + zap.Int64("total_unread", totalUnread)) +} diff --git a/backend/go-service/app/im/service/im_service.go b/backend/go-service/app/im/service/im_service.go index d7a6681..02d422e 100644 --- a/backend/go-service/app/im/service/im_service.go +++ b/backend/go-service/app/im/service/im_service.go @@ -454,6 +454,11 @@ func (s *IMService) GetTotalUnread(ctx context.Context, userID int64) (int64, er return val, err } +// GetPeerUserID 获取单聊会话中对方的用户 ID(供 handler 层调用) +func (s *IMService) GetPeerUserID(ctx context.Context, conversationID, userID int64) (int64, error) { + return s.convDAO.GetPeerUserID(ctx, conversationID, userID) +} + // ====== 内部辅助方法 ====== // getOrCreatePrivateConversation 查找或创建单聊会话 diff --git a/backend/go-service/app/ws/handler.go b/backend/go-service/app/ws/handler.go index c2b1d91..71cb005 100644 --- a/backend/go-service/app/ws/handler.go +++ b/backend/go-service/app/ws/handler.go @@ -21,6 +21,12 @@ type TokenValidator interface { ValidateAccessToken(ctx context.Context, userID int64, clientType, token string) bool } +// OfflineMessagePusher 离线消息推送接口 +// 由 im.handler.OfflinePusher 实现,WebSocket 连接建立后触发推送 +type OfflineMessagePusher interface { + PushOfflineMessages(ctx context.Context, userID int64) +} + var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, @@ -31,11 +37,12 @@ var upgrader = websocket.Upgrader{ // Handler WebSocket 连接处理器 type Handler struct { - hub *ws.Hub - pubsub *ws.PubSub - jwtCfg *config.JWTConfig - onlineService *OnlineService - tokenValidator TokenValidator + hub *ws.Hub + pubsub *ws.PubSub + jwtCfg *config.JWTConfig + onlineService *OnlineService + tokenValidator TokenValidator + offlinePusher OfflineMessagePusher } // NewHandler 创建 WebSocket Handler 实例 @@ -49,6 +56,11 @@ func NewHandler(hub *ws.Hub, pubsub *ws.PubSub, jwtCfg *config.JWTConfig, online } } +// SetOfflinePusher 设置离线消息推送器(由 IM 模块在初始化时注入) +func (h *Handler) SetOfflinePusher(pusher OfflineMessagePusher) { + h.offlinePusher = pusher +} + // Upgrade 处理 WebSocket 升级请求 // GET /ws?token=xxx → JWT 认证 → 升级连接 → 注册 Hub → 订阅 Redis 频道 func (h *Handler) Upgrade(c *gin.Context) { @@ -105,6 +117,10 @@ func (h *Handler) Upgrade(c *gin.Context) { go client.WritePump() go client.ReadPump(h.createReadHandler(claims.UserID)) + + if h.offlinePusher != nil { + go h.offlinePusher.PushOfflineMessages(context.Background(), claims.UserID) + } } // createReadHandler 创建带生命周期管理的消息处理函数