feat(im): Task 4 - WS 事件处理器 + 离线消息推送

EventHandler:
- im.message.send: 发送消息（解析→业务→ACK+推送）
- im.message.recall: 撤回消息
- im.conversation.read: 标记已读
- im.typing: 正在输入通知（转发对方）
- 所有事件通过 Hub.RegisterEvent 注册到路由表

OfflinePusher:
- PushOfflineMessages: 推送未读会话摘要 + 全局未读总数
- 通过 OfflineMessagePusher 接口注入 WS Handler
- WebSocket 连接建立后自动触发离线推送

Made-with: Cursor

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)
+}

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))
+}

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 查找或创建单聊会话

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,
@@ -36,6 +42,7 @@ type Handler struct {
 	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 创建带生命周期管理的消息处理函数