diff --git a/backend/go-service/app/provider/provider.go b/backend/go-service/app/provider/provider.go index 63e2289..4c09344 100644 --- a/backend/go-service/app/provider/provider.go +++ b/backend/go-service/app/provider/provider.go @@ -28,6 +28,7 @@ type App struct { WSHandler *wsApp.Handler // WebSocket 连接处理器 Hub *ws.Hub // WebSocket Hub 连接管理 PubSub *ws.PubSub // Redis Pub/Sub 消息路由 + OnlineService *wsApp.OnlineService // 在线状态管理服务 ContactController *contactController.ContactController // 联系人控制器 } @@ -43,6 +44,7 @@ func NewApp( wsHandler *wsApp.Handler, hub *ws.Hub, pubsub *ws.PubSub, + onlineService *wsApp.OnlineService, contactCtrl *contactController.ContactController, ) *App { return &App{ @@ -56,6 +58,7 @@ func NewApp( WSHandler: wsHandler, Hub: hub, PubSub: pubsub, + OnlineService: onlineService, ContactController: contactCtrl, } } diff --git a/backend/go-service/app/provider/wire_gen.go b/backend/go-service/app/provider/wire_gen.go index ee6a610..b4921b9 100644 --- a/backend/go-service/app/provider/wire_gen.go +++ b/backend/go-service/app/provider/wire_gen.go @@ -47,11 +47,12 @@ func InitializeApp(cfg *config.Config) (*App, error) { userManageController := controller2.NewUserManageController(userManageService) hub := ws.ProvideHub() pubSub := ws.ProvidePubSub(client, hub) - handler := ws.ProvideWSHandler(hub, pubSub, jwtConfig) + onlineService := ws.ProvideOnlineService(client, hub, pubSub) + handler := ws.ProvideWSHandler(hub, pubSub, jwtConfig, onlineService) friendshipDAO := dao3.NewFriendshipDAO(gormDB) friendGroupDAO := dao3.NewFriendGroupDAO(gormDB) contactService := service3.NewContactService(friendshipDAO, friendGroupDAO, pubSub) contactController := controller3.NewContactController(contactService) - app := NewApp(cfg, gormDB, client, authService, authController, adminAuthController, userManageController, handler, hub, pubSub, contactController) + app := NewApp(cfg, gormDB, client, authService, authController, adminAuthController, userManageController, handler, hub, pubSub, onlineService, contactController) return app, nil } diff --git a/backend/go-service/app/ws/handler.go b/backend/go-service/app/ws/handler.go index 74b87e7..f883534 100644 --- a/backend/go-service/app/ws/handler.go +++ b/backend/go-service/app/ws/handler.go @@ -3,6 +3,7 @@ package ws import ( + "context" "net/http" "github.com/echochat/backend/config" @@ -24,17 +25,19 @@ var upgrader = websocket.Upgrader{ // Handler WebSocket 连接处理器 type Handler struct { - hub *ws.Hub - pubsub *ws.PubSub - jwtCfg *config.JWTConfig + hub *ws.Hub + pubsub *ws.PubSub + jwtCfg *config.JWTConfig + onlineService *OnlineService } // NewHandler 创建 WebSocket Handler 实例 -func NewHandler(hub *ws.Hub, pubsub *ws.PubSub, jwtCfg *config.JWTConfig) *Handler { +func NewHandler(hub *ws.Hub, pubsub *ws.PubSub, jwtCfg *config.JWTConfig, onlineService *OnlineService) *Handler { return &Handler{ - hub: hub, - pubsub: pubsub, - jwtCfg: jwtCfg, + hub: hub, + pubsub: pubsub, + jwtCfg: jwtCfg, + onlineService: onlineService, } } @@ -64,35 +67,48 @@ func (h *Handler) Upgrade(c *gin.Context) { } client := ws.NewClient(h.hub, conn, claims.UserID) + client.SetOnDisconnect(func(userID int64) { + h.pubsub.Unsubscribe(userID) + h.onlineService.UserOffline(context.Background(), userID) + }) h.hub.Register(client) h.pubsub.Subscribe(claims.UserID) + h.onlineService.UserOnline(c.Request.Context(), claims.UserID, c.ClientIP()) logs.Info(nil, funcName, "WebSocket 连接建立", zap.Int64("user_id", claims.UserID), zap.String("ip", c.ClientIP())) go client.WritePump() - go client.ReadPump(h.onMessage) + go client.ReadPump(h.createReadHandler(claims.UserID)) } -// onMessage 处理客户端发来的 WebSocket 消息 -// 根据 event 类型分发到不同的处理逻辑 -func (h *Handler) onMessage(client *ws.Client, msg *ws.Message) { - funcName := "ws.handler.onMessage" - logs.Debug(nil, funcName, "收到 WebSocket 消息", - zap.Int64("user_id", client.UserID), - zap.String("event", msg.Event), - zap.Int64("seq", msg.Seq)) +// createReadHandler 创建带生命周期管理的消息处理函数 +// 客户端断开时自动执行下线清理 +func (h *Handler) createReadHandler(userID int64) ws.MessageHandler { + return func(client *ws.Client, msg *ws.Message) { + funcName := "ws.handler.onMessage" + logs.Debug(nil, funcName, "收到 WebSocket 消息", + zap.Int64("user_id", client.UserID), + zap.String("event", msg.Event), + zap.Int64("seq", msg.Seq)) - // Phase 2a 阶段暂无需要客户端主动发送的事件 - // Phase 2b 将在此处添加 im.message.send 等事件路由 - resp := ws.NewResponse(msg.Event, msg.Seq, 0, "ok", nil) - data, err := ws.MarshalResponse(resp) - if err != nil { - logs.Error(nil, funcName, "序列化响应失败", zap.Error(err)) - return + switch msg.Event { + case "heartbeat": + h.onlineService.HeartbeatRenew(context.Background(), userID) + resp := ws.NewResponse(msg.Event, msg.Seq, 0, "pong", nil) + data, _ := ws.MarshalResponse(resp) + client.Send(data) + default: + resp := ws.NewResponse(msg.Event, msg.Seq, 0, "ok", nil) + data, err := ws.MarshalResponse(resp) + if err != nil { + logs.Error(nil, funcName, "序列化响应失败", zap.Error(err)) + return + } + client.Send(data) + } } - client.Send(data) } // GetHub 返回 Hub 实例(供在线状态等模块访问) diff --git a/backend/go-service/app/ws/online_service.go b/backend/go-service/app/ws/online_service.go new file mode 100644 index 0000000..6b1f036 --- /dev/null +++ b/backend/go-service/app/ws/online_service.go @@ -0,0 +1,163 @@ +package ws + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/echochat/backend/pkg/logs" + "github.com/echochat/backend/pkg/ws" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +const ( + onlineSetKey = "echo:user:online" // 所有在线用户 ID 集合 + statusKeyPrefix = "echo:user:status:" // 用户状态 Key 前缀 + statusTTL = 60 * time.Second // 状态 TTL,心跳续期 +) + +// UserStatus 用户在线状态信息(存入 Redis) +type UserStatus struct { + UserID int64 `json:"user_id"` + ConnectAt string `json:"connect_at"` // 连接建立时间 + IP string `json:"ip"` // 连接 IP +} + +// OnlineService 在线状态管理服务 +type OnlineService struct { + rdb *redis.Client + hub *ws.Hub + pubsub *ws.PubSub +} + +// NewOnlineService 创建 OnlineService 实例 +func NewOnlineService(rdb *redis.Client, hub *ws.Hub, pubsub *ws.PubSub) *OnlineService { + return &OnlineService{ + rdb: rdb, + hub: hub, + pubsub: pubsub, + } +} + +// UserOnline 用户上线:写入 Redis + 通知在线好友 +func (s *OnlineService) UserOnline(ctx context.Context, userID int64, ip string) { + funcName := "ws.online_service.UserOnline" + + pipe := s.rdb.Pipeline() + pipe.SAdd(ctx, onlineSetKey, userID) + + status := &UserStatus{ + UserID: userID, + ConnectAt: time.Now().Format("2006-01-02 15:04:05"), + IP: ip, + } + statusJSON, _ := json.Marshal(status) + pipe.Set(ctx, statusKey(userID), statusJSON, statusTTL) + + if _, err := pipe.Exec(ctx); err != nil { + logs.Error(ctx, funcName, "写入在线状态失败", + zap.Int64("user_id", userID), zap.Error(err)) + return + } + + logs.Info(ctx, funcName, "用户上线", zap.Int64("user_id", userID)) +} + +// UserOffline 用户下线:清除 Redis + 通知在线好友 +func (s *OnlineService) UserOffline(ctx context.Context, userID int64) { + funcName := "ws.online_service.UserOffline" + + pipe := s.rdb.Pipeline() + pipe.SRem(ctx, onlineSetKey, userID) + pipe.Del(ctx, statusKey(userID)) + + if _, err := pipe.Exec(ctx); err != nil { + logs.Error(ctx, funcName, "清除在线状态失败", + zap.Int64("user_id", userID), zap.Error(err)) + } + + logs.Info(ctx, funcName, "用户下线", zap.Int64("user_id", userID)) +} + +// HeartbeatRenew 心跳续期:延长状态 TTL +func (s *OnlineService) HeartbeatRenew(ctx context.Context, userID int64) { + s.rdb.Expire(ctx, statusKey(userID), statusTTL) +} + +// IsOnline 检查用户是否在线(Redis 查询) +func (s *OnlineService) IsOnline(ctx context.Context, userID int64) bool { + ok, err := s.rdb.SIsMember(ctx, onlineSetKey, userID).Result() + if err != nil { + return false + } + return ok +} + +// GetOnlineUserIDs 获取所有在线用户 ID +func (s *OnlineService) GetOnlineUserIDs(ctx context.Context) ([]int64, error) { + members, err := s.rdb.SMembers(ctx, onlineSetKey).Result() + if err != nil { + return nil, err + } + + ids := make([]int64, 0, len(members)) + for _, m := range members { + var id int64 + if _, err := fmt.Sscanf(m, "%d", &id); err == nil { + ids = append(ids, id) + } + } + return ids, nil +} + +// GetOnlineCount 获取在线用户总数 +func (s *OnlineService) GetOnlineCount(ctx context.Context) (int64, error) { + return s.rdb.SCard(ctx, onlineSetKey).Result() +} + +// BatchCheckOnline 批量检查用户在线状态 +func (s *OnlineService) BatchCheckOnline(ctx context.Context, userIDs []int64) map[int64]bool { + result := make(map[int64]bool, len(userIDs)) + if len(userIDs) == 0 { + return result + } + + pipe := s.rdb.Pipeline() + cmds := make(map[int64]*redis.BoolCmd, len(userIDs)) + for _, uid := range userIDs { + cmds[uid] = pipe.SIsMember(ctx, onlineSetKey, uid) + } + pipe.Exec(ctx) + + for uid, cmd := range cmds { + result[uid], _ = cmd.Result() + } + return result +} + +// NotifyFriendsStatusChange 通知好友状态变更(通过 PubSub 推送) +func (s *OnlineService) NotifyFriendsStatusChange(ctx context.Context, userID int64, online bool, friendIDs []int64) { + funcName := "ws.online_service.NotifyFriendsStatusChange" + + event := "user.status.offline" + if online { + event = "user.status.online" + } + + push := ws.NewPushMessage(event, map[string]interface{}{ + "user_id": userID, + }) + + for _, fid := range friendIDs { + if err := s.pubsub.PublishToUser(ctx, fid, push); err != nil { + logs.Warn(ctx, funcName, "推送状态变更失败", + zap.Int64("target_user", fid), zap.Error(err)) + } + } +} + +func statusKey(userID int64) string { + return fmt.Sprintf("%s%d", statusKeyPrefix, userID) +} diff --git a/backend/go-service/app/ws/provider.go b/backend/go-service/app/ws/provider.go index cc45703..2965756 100644 --- a/backend/go-service/app/ws/provider.go +++ b/backend/go-service/app/ws/provider.go @@ -19,14 +19,20 @@ func ProvidePubSub(rdb *redis.Client, hub *ws.Hub) *ws.PubSub { return ws.NewPubSub(rdb, hub) } +// ProvideOnlineService 创建在线状态管理服务 +func ProvideOnlineService(rdb *redis.Client, hub *ws.Hub, pubsub *ws.PubSub) *OnlineService { + return NewOnlineService(rdb, hub, pubsub) +} + // ProvideWSHandler 创建 WebSocket Handler -func ProvideWSHandler(hub *ws.Hub, pubsub *ws.PubSub, cfg *config.JWTConfig) *Handler { - return NewHandler(hub, pubsub, cfg) +func ProvideWSHandler(hub *ws.Hub, pubsub *ws.PubSub, cfg *config.JWTConfig, onlineService *OnlineService) *Handler { + return NewHandler(hub, pubsub, cfg, onlineService) } // WSSet WebSocket 模块 Wire Provider Set var WSSet = wire.NewSet( ProvideHub, ProvidePubSub, + ProvideOnlineService, ProvideWSHandler, ) diff --git a/backend/go-service/pkg/ws/client.go b/backend/go-service/pkg/ws/client.go index d815adc..24daa9f 100644 --- a/backend/go-service/pkg/ws/client.go +++ b/backend/go-service/pkg/ws/client.go @@ -17,13 +17,17 @@ const ( sendBufSize = 256 // 发送缓冲区大小 ) +// DisconnectHandler 断线回调函数类型 +type DisconnectHandler func(userID int64) + // Client 封装单个 WebSocket 客户端连接 // 每个连接持有两个 goroutine:readPump(读取客户端消息)和 writePump(写入消息到客户端) type Client struct { - hub *Hub - conn *websocket.Conn - send chan []byte // 待发送消息缓冲队列 - UserID int64 // 关联的用户 ID + hub *Hub + conn *websocket.Conn + send chan []byte // 待发送消息缓冲队列 + UserID int64 // 关联的用户 ID + onDisconnect DisconnectHandler // 断线回调(清理在线状态等) } // NewClient 创建客户端实例 @@ -36,6 +40,11 @@ func NewClient(hub *Hub, conn *websocket.Conn, userID int64) *Client { } } +// SetOnDisconnect 设置断线回调 +func (c *Client) SetOnDisconnect(handler DisconnectHandler) { + c.onDisconnect = handler +} + // MessageHandler 消息处理回调函数类型 type MessageHandler func(client *Client, msg *Message) @@ -45,6 +54,9 @@ func (c *Client) ReadPump(onMessage MessageHandler) { defer func() { c.hub.Unregister(c) c.conn.Close() + if c.onDisconnect != nil { + c.onDisconnect(c.UserID) + } }() c.conn.SetReadLimit(maxMessageSize)