diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 9c74001..e9d7b1b 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -121,7 +121,12 @@ jobs: -e APP_JPUSH_PRODUCTION=true \ -e APP_SENSITIVE_ENABLED=true \ -e APP_REPORT_AUTO_HIDE_THRESHOLD=3 \ - -e "APP_SETUP_SECRET=${{ secrets.SETUP_SECRET }}" \ + -e APP_WEBSOCKET_MODE=standalone \ + -e APP_WEBSOCKET_CLUSTER_INSTANCE_ID= \ + -e APP_WEBSOCKET_CLUSTER_MSG_CHANNEL=ws:msg \ + -e APP_WEBSOCKET_CLUSTER_ONLINE_TTL=60 \ + -e APP_WEBSOCKET_CLUSTER_HEARTBEAT_INTERVAL=20 \ + -e "APP_SETUP_SECRET=${{ secrets.SETUP_SECRET }}" \ ${IMAGE} echo "=== Waiting for startup ===" diff --git a/cmd/server/app.go b/cmd/server/app.go index 35553c1..a3ef7ec 100644 --- a/cmd/server/app.go +++ b/cmd/server/app.go @@ -25,7 +25,7 @@ type App struct { HotRankWorker *service.HotRankWorker GRPCServer *runner.Server Server *http.Server - WsHub *ws.Hub + Publisher ws.MessagePublisher Logger *zap.Logger } @@ -37,7 +37,7 @@ func NewApp( pushService service.PushService, hotRankWorker *service.HotRankWorker, grpcServer *runner.Server, - wsHub *ws.Hub, + publisher ws.MessagePublisher, logger *zap.Logger, ) *App { return &App{ @@ -47,7 +47,7 @@ func NewApp( PushService: pushService, HotRankWorker: hotRankWorker, GRPCServer: grpcServer, - WsHub: wsHub, + Publisher: publisher, Logger: logger, } } @@ -102,9 +102,14 @@ func (a *App) Stop(ctx context.Context) error { a.Logger.Info("Shutting down server") // 1. 关闭所有 WebSocket 连接 - if a.WsHub != nil { + if a.Publisher != nil { a.Logger.Info("Closing WebSocket connections") - a.WsHub.CloseAllConnections() + switch p := a.Publisher.(type) { + case *ws.Bus: + p.Close() + case *ws.Hub: + p.CloseAllConnections() + } } // 3. 停止 HTTP 服务器 diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index 9708f14..4e59a00 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -1,4 +1,4 @@ -// Code generated by Wire. DO NOT EDIT. +// Code generated by Wire. DO NOT EDIT. //go:generate go run -mod=mod github.com/google/wire/cmd/wire //go:build !wireinject @@ -31,12 +31,12 @@ func InitializeApp() (*App, error) { pushRecordRepository := repository.NewPushRecordRepository(db) deviceTokenRepository := repository.NewDeviceTokenRepository(db) messageRepository := repository.NewMessageRepository(db) - hub := wire.ProvideWSHub() + client := wire.ProvideRedisClient(config) + messagePublisher := wire.ProvideWSMessagePublisher(config, client) logger := wire.ProvideLogger() jpushClient := wire.ProvideJPushClient(config, logger) - pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, hub, jpushClient) + pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, messagePublisher, jpushClient) postRepository := repository.NewPostRepository(db) - client := wire.ProvideRedisClient(config) cache := wire.ProvideCache(config, client) systemMessageService := wire.ProvideSystemMessageService(systemNotificationRepository, pushService, userRepository, postRepository, cache) emailClient := wire.ProvideEmailClient(config) @@ -66,10 +66,10 @@ func InitializeApp() (*App, error) { transactionManager := wire.ProvideTransactionManager(db) builtinHooks := wire.ProvideBuiltinHooks(manager) postService := wire.ProvidePostService(postRepository, systemMessageService, postAIService, cache, transactionManager, manager, moderationHooks, builtinHooks, logService) - channelRepository := repository.NewChannelRepository(db) - channelService := wire.ProvideChannelService(channelRepository, cache) postRefRepository := repository.NewPostRefRepository(db) postRefService := wire.ProvidePostRefService(postRefRepository, postRepository) + channelRepository := repository.NewChannelRepository(db) + channelService := wire.ProvideChannelService(channelRepository, cache) postHandler := handler.NewPostHandler(postService, postRefService, userService, channelService) commentService := wire.ProvideCommentService(commentRepository, postRepository, systemMessageService, postAIService, cache, manager, logService) commentHandler := handler.NewCommentHandler(commentService) @@ -78,12 +78,12 @@ func InitializeApp() (*App, error) { return nil, err } uploadService := wire.ProvideUploadService(s3Client, userService) - chatService := wire.ProvideChatService(messageRepository, userRepository, hub, cache, uploadService, pushService) + chatService := wire.ProvideChatService(messageRepository, userRepository, messagePublisher, cache, uploadService, pushService) messageService := wire.ProvideMessageService(messageRepository, cache, uploadService) groupRepository := repository.NewGroupRepository(db) groupJoinRequestRepository := repository.NewGroupJoinRequestRepository(db) - groupService := wire.ProvideGroupService(groupRepository, userRepository, messageRepository, groupJoinRequestRepository, systemNotificationRepository, hub, cache) - messageHandler := wire.ProvideMessageHandler(chatService, messageService, userService, groupService, hub) + groupService := wire.ProvideGroupService(groupRepository, userRepository, messageRepository, groupJoinRequestRepository, systemNotificationRepository, messagePublisher, cache) + messageHandler := wire.ProvideMessageHandler(chatService, messageService, userService, groupService, messagePublisher) notificationRepository := repository.NewNotificationRepository(db) notificationService := wire.ProvideNotificationService(notificationRepository, cache) notificationHandler := handler.NewNotificationHandler(notificationService) @@ -121,14 +121,14 @@ func InitializeApp() (*App, error) { adminDashboardService := wire.ProvideAdminDashboardService(db, userRepository, postRepository, commentRepository, groupRepository, userActivityRepository) adminDashboardHandler := handler.NewAdminDashboardHandler(adminDashboardService) adminLogHandler := handler.NewAdminLogHandler(operationLogService, loginLogService, dataChangeLogService) - qrCodeLoginService := wire.ProvideQRCodeLoginService(client, hub, jwtService, userService, userActivityService, logService) + qrCodeLoginService := wire.ProvideQRCodeLoginService(client, messagePublisher, jwtService, userService, userActivityService, logService) qrCodeHandler := handler.NewQRCodeHandler(qrCodeLoginService) materialSubjectRepository := repository.NewMaterialSubjectRepository(db) materialFileRepository := repository.NewMaterialFileRepository(db) materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository) materialHandler := handler.NewMaterialHandler(materialService) callRepository := repository.NewCallRepository(db) - callService := wire.ProvideCallService(callRepository, hub, config, db) + callService := wire.ProvideCallService(callRepository, messagePublisher, config, db, client) callHandler := handler.NewCallHandler(callService) reportRepository := repository.NewReportRepository(db) reportService := wire.ProvideReportService(reportRepository, postRepository, commentRepository, messageRepository, userRepository, systemNotificationRepository, pushService, transactionManager, operationLogService, config) @@ -143,13 +143,10 @@ func InitializeApp() (*App, error) { adminProfileAuditHandler := handler.NewAdminProfileAuditHandler(userProfileAuditService) setupService := wire.ProvideSetupService(userRepository, roleRepository, casbinService, config) setupHandler := wire.ProvideSetupHandler(setupService) - wsHandler := wire.ProvideWSHandler(hub, chatService, groupService, jwtService, callService, userRepository) - tradeRepository := repository.NewTradeRepository(db) - tradeService := wire.ProvideTradeService(tradeRepository) - tradeHandler := handler.NewTradeHandler(tradeService) - router := ProvideRouter(userRepository, userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, callHandler, reportHandler, adminReportHandler, verificationHandler, adminVerificationHandler, adminProfileAuditHandler, setupHandler, tradeHandler, logService, userActivityService, casbinService, wsHandler) + wsHandler := wire.ProvideWSHandler(messagePublisher, chatService, groupService, jwtService, callService, userRepository) + router := ProvideRouter(userRepository, userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, callHandler, reportHandler, adminReportHandler, verificationHandler, adminVerificationHandler, adminProfileAuditHandler, setupHandler, logService, userActivityService, casbinService, wsHandler) hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, channelRepository, cache) - app := NewApp(config, db, router, pushService, hotRankWorker, server, hub, logger) + app := NewApp(config, db, router, pushService, hotRankWorker, server, messagePublisher, logger) return app, nil } @@ -188,7 +185,6 @@ func ProvideRouter( adminVerificationHandler *handler.AdminVerificationHandler, adminProfileAuditHandler *handler.AdminProfileAuditHandler, setupHandler *handler.SetupHandler, - tradeHandler *handler.TradeHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -226,7 +222,6 @@ func ProvideRouter( AdminVerificationHandler: adminVerificationHandler, AdminProfileAuditHandler: adminProfileAuditHandler, SetupHandler: setupHandler, - TradeHandler: tradeHandler, LogService: logService, ActivityService: activityService, CasbinService: casbinService, diff --git a/configs/config.yaml b/configs/config.yaml index 9d4f8b3..7b6826e 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -277,3 +277,16 @@ jpush: # 设置后可使用 POST /api/v1/admin/setup-super-admin 接口初始化第一位超级管理员 # 该接口只能使用一次，一旦系统中存在超级管理员，此接口将永久拒绝请求 setup_secret: "" + +# WebSocket 配置 +# 环境变量: +# APP_WEBSOCKET_MODE, APP_WEBSOCKET_CLUSTER_INSTANCE_ID +# APP_WEBSOCKET_CLUSTER_MSG_CHANNEL, APP_WEBSOCKET_CLUSTER_ONLINE_TTL +# APP_WEBSOCKET_CLUSTER_HEARTBEAT_INTERVAL +websocket: + mode: standalone # standalone 或 cluster + cluster: + instance_id: "" # 留空自动生成 UUID + msg_channel: "ws:msg" # Redis Pub/Sub 频道 + online_ttl: 60 # 在线状态 TTL（秒） + heartbeat_interval: 20 # 心跳间隔（秒） diff --git a/internal/config/config.go b/internal/config/config.go index 283a4fe..5abbbcf 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -32,6 +32,7 @@ type Config struct { Sensitive SensitiveConfig `mapstructure:"sensitive"` JPush JPushConfig `mapstructure:"jpush"` SetupSecret string `mapstructure:"setup_secret"` + WebSocket WSConfig `mapstructure:"websocket"` } // Load 加载配置文件 @@ -173,6 +174,12 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("jpush.master_secret", "") viper.SetDefault("jpush.production", false) viper.SetDefault("setup_secret", "") + // WebSocket 默认值 + viper.SetDefault("websocket.mode", "standalone") + viper.SetDefault("websocket.cluster.instance_id", "") + viper.SetDefault("websocket.cluster.msg_channel", "ws:msg") + viper.SetDefault("websocket.cluster.online_ttl", 60) + viper.SetDefault("websocket.cluster.heartbeat_interval", 20) if err := viper.ReadInConfig(); err != nil { return nil, fmt.Errorf("failed to read config: %w", err) diff --git a/internal/config/websocket.go b/internal/config/websocket.go new file mode 100644 index 0000000..c76de83 --- /dev/null +++ b/internal/config/websocket.go @@ -0,0 +1,15 @@ +package config + +// WSConfig WebSocket 配置 +type WSConfig struct { + Mode string `mapstructure:"mode"` // standalone | cluster + Cluster WSClusterConfig `mapstructure:"cluster"` +} + +// WSClusterConfig WebSocket 集群配置 +type WSClusterConfig struct { + InstanceID string `mapstructure:"instance_id"` + MsgChannel string `mapstructure:"msg_channel"` + OnlineTTL int `mapstructure:"online_ttl"` + HeartbeatInterval int `mapstructure:"heartbeat_interval"` +} diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 4654e13..ac1da27 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -90,17 +90,17 @@ type MessageHandler struct { messageService *service.MessageService userService service.UserService groupService service.GroupService - wsHub *ws.Hub + wsPublisher ws.MessagePublisher } // NewMessageHandler 创建消息处理器 -func NewMessageHandler(chatService service.ChatService, messageService *service.MessageService, userService service.UserService, groupService service.GroupService, wsHub *ws.Hub) *MessageHandler { +func NewMessageHandler(chatService service.ChatService, messageService *service.MessageService, userService service.UserService, groupService service.GroupService, wsPublisher ws.MessagePublisher) *MessageHandler { return &MessageHandler{ chatService: chatService, messageService: messageService, userService: userService, groupService: groupService, - wsHub: wsHub, + wsPublisher: wsPublisher, } } diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 0abdab9..32e1a11 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -67,7 +67,7 @@ func isAllowedWebSocketOrigin(origin string) bool { // WSHandler WebSocket处理器 type WSHandler struct { - wsHub *ws.Hub + publisher ws.MessagePublisher chatService service.ChatService groupService service.GroupService jwtService *service.JWTService @@ -76,9 +76,21 @@ type WSHandler struct { clientSeq uint64 } +// hub 获取底层 Hub（用于 Register/Unregister/AckMessage 等本地操作） +func (h *WSHandler) hub() *ws.Hub { + switch p := h.publisher.(type) { + case *ws.Bus: + return p.Hub() + case *ws.Hub: + return p + default: + return nil + } +} + // NewWSHandler 创建WebSocket处理器 func NewWSHandler( - wsHub *ws.Hub, + publisher ws.MessagePublisher, chatService service.ChatService, groupService service.GroupService, jwtService *service.JWTService, @@ -86,7 +98,7 @@ func NewWSHandler( userRepo repository.UserRepository, ) *WSHandler { return &WSHandler{ - wsHub: wsHub, + publisher: publisher, chatService: chatService, groupService: groupService, jwtService: jwtService, @@ -148,7 +160,7 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { } // 4. 注册客户端 - regErr := h.wsHub.Register(client) + regErr := h.hub().Register(client) if regErr != nil { zap.L().Warn("WebSocket registration rejected", zap.String("user_id", userID), @@ -159,7 +171,7 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { conn.Close() return } - defer h.wsHub.Unregister(client) + defer h.hub().Unregister(client) zap.L().Info("WebSocket client connected", zap.String("user_id", userID), @@ -168,7 +180,7 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { // 5. 提示客户端进行 seq 同步 syncMsg := ws.ResponseMessage{ - EventID: h.wsHub.NextID(), + EventID: h.publisher.NextID(), Type: "sync_required", TS: time.Now().UnixMilli(), } @@ -234,7 +246,7 @@ func (h *WSHandler) readPump(conn *websocket.Conn, client *ws.Client) { // 解析消息 var msg ws.Message if err := json.Unmarshal(message, &msg); err != nil { - h.wsHub.SendError(client, "parse_error", "invalid message format") + h.publisher.SendError(client, "parse_error", "invalid message format") continue } @@ -338,14 +350,14 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) { case "call_mute": h.handleCallMute(ctx, client, msg.Payload) default: - h.wsHub.SendError(client, "unknown_type", fmt.Sprintf("unknown message type: %s", msg.Type)) + h.publisher.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(), + EventID: h.publisher.NextID(), Type: "pong", TS: time.Now().UnixMilli(), } @@ -371,24 +383,24 @@ func (h *WSHandler) handleChat(ctx context.Context, client *ws.Client, payload j } if err := json.Unmarshal(payload, &req); err != nil { - h.wsHub.SendError(client, "parse_error", "invalid chat message format") + h.publisher.SendError(client, "parse_error", "invalid chat message format") return } if req.ConversationID == "" { - h.wsHub.SendError(client, "invalid_params", "conversation_id is required") + h.publisher.SendError(client, "invalid_params", "conversation_id is required") return } if len(req.Segments) == 0 { - h.wsHub.SendError(client, "invalid_params", "segments is required") + h.publisher.SendError(client, "invalid_params", "segments is required") return } // 发送消息 message, err := h.chatService.SendMessage(ctx, client.UserID, req.ConversationID, req.Segments, req.ReplyToID, req.ClientMsgID) if err != nil { - h.wsHub.SendError(client, "send_failed", err.Error()) + h.publisher.SendError(client, "send_failed", err.Error()) return } @@ -410,7 +422,7 @@ func (h *WSHandler) handleChat(ctx context.Context, client *ws.Client, payload j } msg := ws.ResponseMessage{ - EventID: h.wsHub.NextID(), + EventID: h.publisher.NextID(), Type: "message_sent", TS: time.Now().UnixMilli(), Payload: resp, @@ -478,24 +490,24 @@ func (h *WSHandler) handleRecall(ctx context.Context, client *ws.Client, payload } if err := json.Unmarshal(payload, &req); err != nil { - h.wsHub.SendError(client, "parse_error", "invalid recall message format") + h.publisher.SendError(client, "parse_error", "invalid recall message format") return } if req.MessageID == "" { - h.wsHub.SendError(client, "invalid_params", "message_id is required") + h.publisher.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()) + h.publisher.SendError(client, "recall_failed", err.Error()) return } // 发送撤回成功响应 resp := ws.ResponseMessage{ - EventID: h.wsHub.NextID(), + EventID: h.publisher.NextID(), Type: "message_recalled", TS: time.Now().UnixMilli(), Payload: map[string]string{"message_id": req.MessageID}, @@ -517,18 +529,18 @@ func (h *WSHandler) handleAck(client *ws.Client, payload json.RawMessage) { if err := json.Unmarshal(payload, &req); err != nil || req.MessageID == "" { return } - h.wsHub.AckMessage(client.UserID, req.MessageID) + h.hub().AckMessage(client.UserID, req.MessageID) } // isVerified 检查用户是否已通过身份认证 func (h *WSHandler) isVerified(ctx context.Context, client *ws.Client) bool { user, err := h.userRepo.GetByID(client.UserID) if err != nil { - h.wsHub.SendError(client, "internal_error", "获取用户信息失败") + h.publisher.SendError(client, "internal_error", "获取用户信息失败") return false } if user.VerificationStatus != model.VerificationStatusApproved { - h.wsHub.SendError(client, "VERIFICATION_REQUIRED", "请先完成身份认证") + h.publisher.SendError(client, "VERIFICATION_REQUIRED", "请先完成身份认证") return false } return true @@ -548,11 +560,11 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay MediaType string `json:"call_type"` // voice 或 video } if err := json.Unmarshal(payload, &req); err != nil { - h.wsHub.SendError(client, "parse_error", "invalid call_invite format") + h.publisher.SendError(client, "parse_error", "invalid call_invite format") return } if req.CalleeID == "" || req.ConversationID == "" { - h.wsHub.SendError(client, "invalid_params", "callee_id and conversation_id are required") + h.publisher.SendError(client, "invalid_params", "callee_id and conversation_id are required") return } @@ -570,16 +582,16 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay zap.Error(err), ) if errors.Is(err, apperrors.ErrCallInProgress) { - h.wsHub.SendError(client, "call_in_progress", err.Error()) + h.publisher.SendError(client, "call_in_progress", err.Error()) return } - h.wsHub.SendError(client, "call_invite_failed", err.Error()) + h.publisher.SendError(client, "call_invite_failed", err.Error()) return } // 返回给呼叫方的响应 resp := ws.ResponseMessage{ - EventID: h.wsHub.NextID(), + EventID: h.publisher.NextID(), Type: "call_invited", TS: time.Now().UnixMilli(), Payload: map[string]any{ @@ -602,7 +614,7 @@ func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, pay CallID string `json:"call_id"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { - h.wsHub.SendError(client, "invalid_params", "call_id is required") + h.publisher.SendError(client, "invalid_params", "call_id is required") return } @@ -610,15 +622,15 @@ func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, pay if err != nil { // === 区分已接听错误 === if errors.Is(err, apperrors.ErrCallAlreadyAnswered) { - h.wsHub.SendError(client, "call_already_answered", "通话已被其他设备接听") + h.publisher.SendError(client, "call_already_answered", "通话已被其他设备接听") return } - h.wsHub.SendError(client, "call_accept_failed", err.Error()) + h.publisher.SendError(client, "call_accept_failed", err.Error()) return } resp := ws.ResponseMessage{ - EventID: h.wsHub.NextID(), + EventID: h.publisher.NextID(), Type: "call_accepted", TS: time.Now().UnixMilli(), Payload: map[string]any{ @@ -639,12 +651,12 @@ func (h *WSHandler) handleCallReject(ctx context.Context, client *ws.Client, pay CallID string `json:"call_id"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { - h.wsHub.SendError(client, "invalid_params", "call_id is required") + h.publisher.SendError(client, "invalid_params", "call_id is required") return } if err := h.callService.Reject(ctx, req.CallID, client.UserID); err != nil { - h.wsHub.SendError(client, "call_reject_failed", err.Error()) + h.publisher.SendError(client, "call_reject_failed", err.Error()) } } @@ -654,12 +666,12 @@ func (h *WSHandler) handleCallBusy(ctx context.Context, client *ws.Client, paylo CallID string `json:"call_id"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { - h.wsHub.SendError(client, "invalid_params", "call_id is required") + h.publisher.SendError(client, "invalid_params", "call_id is required") return } if err := h.callService.Busy(ctx, req.CallID, client.UserID); err != nil { - h.wsHub.SendError(client, "call_busy_failed", err.Error()) + h.publisher.SendError(client, "call_busy_failed", err.Error()) } } @@ -671,7 +683,7 @@ func (h *WSHandler) handleCallSDP(ctx context.Context, client *ws.Client, payloa SDP json.RawMessage `json:"sdp"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" || req.SDPType == "" { - h.wsHub.SendError(client, "invalid_params", "call_id and sdp_type are required") + h.publisher.SendError(client, "invalid_params", "call_id and sdp_type are required") return } @@ -680,7 +692,7 @@ func (h *WSHandler) handleCallSDP(ctx context.Context, client *ws.Client, payloa "sdp": json.RawMessage(req.SDP), }) if err := h.callService.RelaySignal(ctx, req.CallID, client.UserID, "call_sdp", signalPayload); err != nil { - h.wsHub.SendError(client, "call_sdp_failed", err.Error()) + h.publisher.SendError(client, "call_sdp_failed", err.Error()) } } @@ -691,7 +703,7 @@ func (h *WSHandler) handleCallICE(ctx context.Context, client *ws.Client, payloa Candidate json.RawMessage `json:"candidate"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { - h.wsHub.SendError(client, "invalid_params", "call_id is required") + h.publisher.SendError(client, "invalid_params", "call_id is required") return } @@ -699,7 +711,7 @@ func (h *WSHandler) handleCallICE(ctx context.Context, client *ws.Client, payloa "candidate": json.RawMessage(req.Candidate), }) if err := h.callService.RelaySignal(ctx, req.CallID, client.UserID, "call_ice", signalPayload); err != nil { - h.wsHub.SendError(client, "call_ice_failed", err.Error()) + h.publisher.SendError(client, "call_ice_failed", err.Error()) } } @@ -710,18 +722,18 @@ func (h *WSHandler) handleCallEnd(ctx context.Context, client *ws.Client, payloa Reason string `json:"reason"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { - h.wsHub.SendError(client, "invalid_params", "call_id is required") + h.publisher.SendError(client, "invalid_params", "call_id is required") return } call, err := h.callService.End(ctx, req.CallID, client.UserID, req.Reason) if err != nil { - h.wsHub.SendError(client, "call_end_failed", err.Error()) + h.publisher.SendError(client, "call_end_failed", err.Error()) return } resp := ws.ResponseMessage{ - EventID: h.wsHub.NextID(), + EventID: h.publisher.NextID(), Type: "call_ended", TS: time.Now().UnixMilli(), Payload: map[string]any{ @@ -743,11 +755,11 @@ func (h *WSHandler) handleCallMute(ctx context.Context, client *ws.Client, paylo Muted bool `json:"muted"` } if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { - h.wsHub.SendError(client, "invalid_params", "call_id is required") + h.publisher.SendError(client, "invalid_params", "call_id is required") return } if err := h.callService.SetMuted(ctx, req.CallID, client.UserID, req.Muted); err != nil { - h.wsHub.SendError(client, "call_mute_failed", err.Error()) + h.publisher.SendError(client, "call_mute_failed", err.Error()) } } diff --git a/internal/pkg/ws/bus.go b/internal/pkg/ws/bus.go new file mode 100644 index 0000000..832a657 --- /dev/null +++ b/internal/pkg/ws/bus.go @@ -0,0 +1,395 @@ +package ws + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/google/uuid" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// busMsg Redis Pub/Sub 消息格式 +type busMsg struct { + InstanceID string `json:"instance_id"` + MsgType string `json:"msg_type"` // "user_msg" | "broadcast" + TargetUsers []string `json:"target_users"` + EventType string `json:"event_type"` + EventID uint64 `json:"event_id"` + TS int64 `json:"ts"` + Payload json.RawMessage `json:"payload"` +} + +// WSClusterConfig WebSocket 集群配置 +type WSClusterConfig struct { + InstanceID string + MsgChannel string + OnlineTTL int + HeartbeatInterval int +} + +// Bus 组合 Hub + Redis Pub/Sub + OnlineTracker，实现跨实例消息路由 +type Bus struct { + hub *Hub + rdb *redis.Client + tracker *OnlineTracker + + instanceID string + channel string + sub *redis.PubSub + cancel context.CancelFunc + + mu sync.Mutex + closed bool +} + +// NewBus 创建 Bus 实例 +// 如果 rdb 为 nil（standalone 模式或 Redis 不可用），Bus 退化为纯本地 Hub 代理 +func NewBus(hub *Hub, rdb *redis.Client, cfg *WSClusterConfig) *Bus { + instanceID := cfg.InstanceID + if instanceID == "" { + instanceID = uuid.New().String() + } + + onlineTTL := time.Duration(cfg.OnlineTTL) * time.Second + heartbeatInterval := time.Duration(cfg.HeartbeatInterval) * time.Second + if heartbeatInterval <= 0 { + heartbeatInterval = 20 * time.Second + } + if onlineTTL <= 0 { + onlineTTL = 60 * time.Second + } + + tracker := NewOnlineTracker(rdb, instanceID, onlineTTL, heartbeatInterval) + + return &Bus{ + hub: hub, + rdb: rdb, + tracker: tracker, + instanceID: instanceID, + channel: cfg.MsgChannel, + } +} + +// Hub 获取底层 Hub（供 WebSocket handler 使用 Register/Unregister 等本地操作） +func (b *Bus) Hub() *Hub { + return b.hub +} + +// InstanceID 获取实例 ID +func (b *Bus) InstanceID() string { + return b.instanceID +} + +// Start 启动 Bus 的 Redis 订阅和心跳 +func (b *Bus) Start() error { + if b.rdb == nil { + zap.L().Info("WebSocket Bus: no Redis client, running in standalone mode") + return nil + } + + // 启动心跳 + b.tracker.StartHeartbeat() + + // 订阅 Redis channel + ctx, cancel := context.WithCancel(context.Background()) + b.cancel = cancel + + sub := b.rdb.Subscribe(ctx, b.channel) + b.sub = sub + + // 确保订阅已建立 + if _, err := sub.Receive(ctx); err != nil { + cancel() + return err + } + + go b.subscribeLoop(ctx, sub) + + zap.L().Info("WebSocket Bus started", + zap.String("instance_id", b.instanceID), + zap.String("channel", b.channel), + ) + return nil +} + +// Register 注册客户端连接（代理 Hub.Register + tracker） +func (b *Bus) Register(client *Client) error { + if err := b.hub.Register(client); err != nil { + return err + } + b.tracker.UserConnected(client.UserID) + return nil +} + +// Unregister 注销客户端连接（代理 Hub.Unregister + tracker） +func (b *Bus) Unregister(client *Client) { + b.hub.Unregister(client) + b.tracker.UserDisconnected(client.UserID) +} + +// Close 关闭 Bus：关闭订阅，清理在线状态 +func (b *Bus) Close() error { + b.mu.Lock() + defer b.mu.Unlock() + if b.closed { + return nil + } + b.closed = true + + if b.cancel != nil { + b.cancel() + } + if b.sub != nil { + b.sub.Close() + } + b.tracker.Stop() + b.hub.CloseAllConnections() + + zap.L().Info("WebSocket Bus closed", zap.String("instance_id", b.instanceID)) + return nil +} + +// --- MessagePublisher 接口实现 --- + +// PublishToUser 向指定用户发送事件（本地 + 远程） +func (b *Bus) PublishToUser(userID string, eventType string, payload any) Event { + ev := b.hub.PublishToUser(userID, eventType, payload) + b.publishToRemote([]string{userID}, "user_msg", ev, payload) + return ev +} + +// PublishToUsers 向多个用户发送事件 +func (b *Bus) PublishToUsers(userIDs []string, eventType string, payload any) { + b.hub.PublishToUsers(userIDs, eventType, payload) + b.publishToRemote(userIDs, "user_msg", Event{ + Type: eventType, + TS: time.Now().UnixMilli(), + }, payload) +} + +// PublishToUsersConcurrent 并发推送消息给多个用户 +func (b *Bus) PublishToUsersConcurrent(userIDs []string, eventType string, payload any) { + b.hub.PublishToUsersConcurrent(userIDs, eventType, payload) + b.publishToRemote(userIDs, "user_msg", Event{ + Type: eventType, + TS: time.Now().UnixMilli(), + }, payload) +} + +// PublishToUserOnline 仅在用户在线时发送事件 +func (b *Bus) PublishToUserOnline(userID string, eventType string, payload any) bool { + delivered := b.hub.PublishToUserOnline(userID, eventType, payload) + if !delivered { + // 本地不在线，尝试远程投递 + ev := Event{ + Type: eventType, + TS: time.Now().UnixMilli(), + } + b.publishToRemote([]string{userID}, "user_msg", ev, payload) + } + return delivered +} + +// PublishToUserOnlineReliable 可靠地发送事件给在线用户 +func (b *Bus) PublishToUserOnlineReliable(userID string, eventType string, payload any) bool { + delivered := b.hub.PublishToUserOnlineReliable(userID, eventType, payload) + if !delivered { + ev := Event{ + Type: eventType, + TS: time.Now().UnixMilli(), + } + b.publishToRemote([]string{userID}, "user_msg", ev, payload) + } + return delivered +} + +// Broadcast 广播消息给所有连接 +func (b *Bus) Broadcast(eventType string, payload any) { + b.hub.Broadcast(eventType, payload) + b.publishToRemote(nil, "broadcast", Event{ + Type: eventType, + TS: time.Now().UnixMilli(), + }, payload) +} + +// HasClients 检查用户是否在线（本地优先，然后查 Redis） +func (b *Bus) HasClients(userID string) bool { + if b.hub.HasClients(userID) { + return true + } + return b.tracker.IsOnline(userID) +} + +// FilterOnline 批量检查用户在线状态 +func (b *Bus) FilterOnline(userIDs []string) (online, offline []string) { + localOnline, offline := b.hub.FilterOnline(userIDs) + if b.rdb == nil { + return localOnline, offline + } + // 对 offline 的列表，查 Redis + remoteOnline, stillOffline := b.tracker.FilterOnline(offline) + online = append(localOnline, remoteOnline...) + return online, stillOffline +} + +// SendError 向客户端发送错误消息（本地操作） +func (b *Bus) SendError(client *Client, code string, message string) { + b.hub.SendError(client, code, message) +} + +// NextID 获取下一个事件 ID +func (b *Bus) NextID() uint64 { + return b.hub.NextID() +} + +// OnDisconnect 注册断开连接事件处理器 +func (b *Bus) OnDisconnect(handler DisconnectHandler) { + b.hub.OnDisconnect(handler) +} + +// OnConnect 注册连接事件处理器 +func (b *Bus) OnConnect(handler ConnectHandler) { + b.hub.OnConnect(handler) +} + +// --- 内部方法 --- + +// publishToRemote 发布消息到 Redis channel +func (b *Bus) publishToRemote(targetUsers []string, msgType string, ev Event, payload any) { + if b.rdb == nil { + return + } + + payloadBytes, err := json.Marshal(payload) + if err != nil { + zap.L().Error("Bus: failed to marshal payload for remote publish", + zap.String("event_type", ev.Type), + zap.Error(err), + ) + return + } + + msg := busMsg{ + InstanceID: b.instanceID, + MsgType: msgType, + TargetUsers: targetUsers, + EventType: ev.Type, + EventID: ev.ID, + TS: ev.TS, + Payload: payloadBytes, + } + + data, err := json.Marshal(msg) + if err != nil { + zap.L().Error("Bus: failed to marshal busMsg", zap.Error(err)) + return + } + + if err := b.rdb.Publish(context.Background(), b.channel, data).Err(); err != nil { + zap.L().Error("Bus: failed to publish to Redis", + zap.String("channel", b.channel), + zap.Error(err), + ) + } +} + +// subscribeLoop 处理 Redis 订阅消息 +func (b *Bus) subscribeLoop(ctx context.Context, sub *redis.PubSub) { + ch := sub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + b.handleRemoteMessage(msg) + } + } +} + +// handleRemoteMessage 处理从 Redis 收到的远程消息 +func (b *Bus) handleRemoteMessage(msg *redis.Message) { + var bm busMsg + if err := json.Unmarshal([]byte(msg.Payload), &bm); err != nil { + zap.L().Error("Bus: failed to unmarshal remote message", zap.Error(err)) + return + } + + // 过滤自身发送的消息 + if bm.InstanceID == b.instanceID { + return + } + + switch bm.MsgType { + case "user_msg": + b.handleRemoteUserMsg(bm) + case "broadcast": + b.handleRemoteBroadcast(bm) + default: + zap.L().Warn("Bus: unknown msg_type from remote", + zap.String("msg_type", bm.MsgType), + ) + } +} + +// handleRemoteUserMsg 处理远程定向消息 +func (b *Bus) handleRemoteUserMsg(bm busMsg) { + // 构造 ResponseMessage 用于本地投递 + respMsg := ResponseMessage{ + EventID: bm.EventID, + Type: bm.EventType, + TS: bm.TS, + } + + // 反序列化 payload + var payload any + if err := json.Unmarshal(bm.Payload, &payload); err != nil { + zap.L().Error("Bus: failed to unmarshal remote payload", zap.Error(err)) + return + } + respMsg.Payload = payload + + data, err := json.Marshal(respMsg) + if err != nil { + zap.L().Error("Bus: failed to marshal response message", zap.Error(err)) + return + } + + b.hub.deliverToUsers(bm.TargetUsers, data) +} + +// handleRemoteBroadcast 处理远程广播消息 +func (b *Bus) handleRemoteBroadcast(bm busMsg) { + b.hub.Broadcast(bm.EventType, bm.Payload) +} + +// deliverToUsers 向指定用户列表投递预序列化消息 +// 这是 Hub 批量投递的内部方法，Bus 需要调用它来投递远程消息 +func (h *Hub) deliverToUsers(userIDs []string, data []byte) { + for _, uid := range userIDs { + h.mu.RLock() + targets := make([]*Client, 0, len(h.clients[uid])) + for _, c := range h.clients[uid] { + targets = append(targets, c) + } + h.mu.RUnlock() + + for _, c := range targets { + select { + case <-c.Quit: + case c.Send <- data: + default: + zap.L().Warn("WebSocket client buffer full, closing slow client", + zap.String("user_id", uid), + zap.Uint64("client_id", c.ID), + ) + close(c.Quit) + } + } + } +} \ No newline at end of file diff --git a/internal/pkg/ws/online_tracker.go b/internal/pkg/ws/online_tracker.go new file mode 100644 index 0000000..28ffd35 --- /dev/null +++ b/internal/pkg/ws/online_tracker.go @@ -0,0 +1,257 @@ +package ws + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// OnlineTracker 分布式在线状态追踪 +// 使用 Redis Hash 存储各实例的用户连接数，支持跨实例在线查询 +type OnlineTracker struct { + rdb *redis.Client + instanceID string + localUsers sync.Map // 本实例在线用户集合: userID -> connCount + onlineTTL time.Duration + heartbeatInterval time.Duration + cancel context.CancelFunc +} + +// NewOnlineTracker 创建在线状态追踪器 +func NewOnlineTracker(rdb *redis.Client, instanceID string, onlineTTL, heartbeatInterval time.Duration) *OnlineTracker { + return &OnlineTracker{ + rdb: rdb, + instanceID: instanceID, + onlineTTL: onlineTTL, + heartbeatInterval: heartbeatInterval, + } +} + +// UserConnected 用户连接，更新 Redis 和本地在线状态 +func (t *OnlineTracker) UserConnected(userID string) { + if t.rdb == nil { + return + } + ctx := context.Background() + key := fmt.Sprintf("online:%s", userID) + + // 本地计数 + count := int64(0) + if v, ok := t.localUsers.Load(userID); ok { + count = v.(int64) + } + count++ + t.localUsers.Store(userID, count) + + // Redis: HINCRBY + EXPIRE + pipe := t.rdb.Pipeline() + pipe.HIncrBy(ctx, key, t.instanceID, 1) + pipe.Expire(ctx, key, t.onlineTTL) + if _, err := pipe.Exec(ctx); err != nil { + zap.L().Error("Failed to update online status on connect", + zap.String("user_id", userID), + zap.Error(err), + ) + } +} + +// UserDisconnected 用户断开连接，更新 Redis 和本地在线状态 +// 返回该用户剩余的连接数 +func (t *OnlineTracker) UserDisconnected(userID string) int { + if t.rdb == nil { + if v, ok := t.localUsers.Load(userID); ok { + count := v.(int64) - 1 + if count <= 0 { + t.localUsers.Delete(userID) + return 0 + } + t.localUsers.Store(userID, count) + return int(count) + } + return 0 + } + + // 本地计数递减 + count := int64(0) + if v, ok := t.localUsers.Load(userID); ok { + count = v.(int64) - 1 + } + + if count <= 0 { + t.localUsers.Delete(userID) + // Redis: 删除该实例的 field + ctx := context.Background() + key := fmt.Sprintf("online:%s", userID) + t.rdb.HDel(ctx, key, t.instanceID) + // 检查 key 是否为空，如果为空则删除 + exists, _ := t.rdb.HLen(ctx, key).Result() + if exists == 0 { + t.rdb.Del(ctx, key) + } + return 0 + } + + t.localUsers.Store(userID, count) + + // Redis: HINCRBY -1 + ctx := context.Background() + key := fmt.Sprintf("online:%s", userID) + newVal, err := t.rdb.HIncrBy(ctx, key, t.instanceID, -1).Result() + if err != nil { + zap.L().Error("Failed to update online status on disconnect", + zap.String("user_id", userID), + zap.Error(err), + ) + return int(count) + } + + if newVal <= 0 { + t.rdb.HDel(ctx, key, t.instanceID) + exists, _ := t.rdb.HLen(ctx, key).Result() + if exists == 0 { + t.rdb.Del(ctx, key) + } + } + + return int(count) +} + +// IsOnline 检查用户是否在线（查询 Redis） +func (t *OnlineTracker) IsOnline(userID string) bool { + if t.rdb == nil { + _, ok := t.localUsers.Load(userID) + return ok + } + ctx := context.Background() + key := fmt.Sprintf("online:%s", userID) + exists, err := t.rdb.Exists(ctx, key).Result() + if err != nil { + zap.L().Error("Failed to check online status", + zap.String("user_id", userID), + zap.Error(err), + ) + // 降级：检查本地 + _, ok := t.localUsers.Load(userID) + return ok + } + return exists > 0 +} + +// FilterOnline 批量检查用户在线状态 +func (t *OnlineTracker) FilterOnline(userIDs []string) (online, offline []string) { + if t.rdb == nil { + // 降级：只检查本地 + for _, uid := range userIDs { + if _, ok := t.localUsers.Load(uid); ok { + online = append(online, uid) + } else { + offline = append(offline, uid) + } + } + return + } + + ctx := context.Background() + pipe := t.rdb.Pipeline() + cmds := make([]*redis.IntCmd, len(userIDs)) + for i, uid := range userIDs { + key := fmt.Sprintf("online:%s", uid) + cmds[i] = pipe.Exists(ctx, key) + } + if _, err := pipe.Exec(ctx); err != nil { + zap.L().Error("Failed to batch check online status", zap.Error(err)) + // 降级 + for _, uid := range userIDs { + if _, ok := t.localUsers.Load(uid); ok { + online = append(online, uid) + } else { + offline = append(offline, uid) + } + } + return + } + + for i, uid := range userIDs { + if cmds[i].Val() > 0 { + online = append(online, uid) + } else { + offline = append(offline, uid) + } + } + return +} + +// StartHeartbeat 启动心跳续期 +func (t *OnlineTracker) StartHeartbeat() { + if t.rdb == nil { + return + } + ctx, cancel := context.WithCancel(context.Background()) + t.cancel = cancel + + go func() { + ticker := time.NewTicker(t.heartbeatInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + t.heartbeat() + } + } + }() +} + +func (t *OnlineTracker) heartbeat() { + ctx := context.Background() + pipe := t.rdb.Pipeline() + count := 0 + + t.localUsers.Range(func(key, value any) bool { + userID := key.(string) + onlineKey := fmt.Sprintf("online:%s", userID) + pipe.Expire(ctx, onlineKey, t.onlineTTL) + count++ + return true + }) + + if count > 0 { + if _, err := pipe.Exec(ctx); err != nil { + zap.L().Error("Failed to heartbeat online keys", zap.Error(err)) + } + } +} + +// Stop 停止心跳并清理本实例的在线状态 +func (t *OnlineTracker) Stop() { + if t.cancel != nil { + t.cancel() + } + + if t.rdb == nil { + return + } + + ctx := context.Background() + pipe := t.rdb.Pipeline() + + // 清理本实例在所有在线 key 中的 field + t.localUsers.Range(func(key, value any) bool { + userID := key.(string) + onlineKey := fmt.Sprintf("online:%s", userID) + pipe.HDel(ctx, onlineKey, t.instanceID) + // 删除后检查是否为空 + pipe.HLen(ctx, onlineKey) + return true + }) + + if _, err := pipe.Exec(ctx); err != nil { + zap.L().Error("Failed to cleanup online status on stop", zap.Error(err)) + } +} \ No newline at end of file diff --git a/internal/pkg/ws/publisher.go b/internal/pkg/ws/publisher.go new file mode 100644 index 0000000..9c87f2a --- /dev/null +++ b/internal/pkg/ws/publisher.go @@ -0,0 +1,23 @@ +package ws + +// MessagePublisher 定义消息发布接口 +// ws.Hub 隐式满足此接口，ws.Bus 组合 Hub 并扩展跨实例能力 +type MessagePublisher interface { + // 消息发布 + PublishToUser(userID string, eventType string, payload any) Event + PublishToUsers(userIDs []string, eventType string, payload any) + PublishToUsersConcurrent(userIDs []string, eventType string, payload any) + PublishToUserOnline(userID string, eventType string, payload any) bool + PublishToUserOnlineReliable(userID string, eventType string, payload any) bool + Broadcast(eventType string, payload any) + + // 在线状态 + HasClients(userID string) bool + FilterOnline(userIDs []string) (online, offline []string) + + // 本地操作（不经过 Redis） + SendError(client *Client, code string, message string) + NextID() uint64 + OnDisconnect(handler DisconnectHandler) + OnConnect(handler ConnectHandler) +} \ No newline at end of file diff --git a/internal/service/account_cleanup_service.go b/internal/service/account_cleanup_service.go index 199e306..1b63b33 100644 --- a/internal/service/account_cleanup_service.go +++ b/internal/service/account_cleanup_service.go @@ -2,12 +2,12 @@ package service import ( "context" - "log" "time" "with_you/internal/model" "with_you/internal/repository" + "go.uber.org/zap" "gorm.io/gorm" ) @@ -50,13 +50,18 @@ func (s *accountCleanupServiceImpl) CleanupExpiredAccounts(ctx context.Context) var deletedCount int64 for _, user := range users { if err := s.anonymizeAndDeleteUser(ctx, user); err != nil { - log.Printf("[AccountCleanup] failed to cleanup user %s: %v", user.ID, err) + zap.L().Error("Failed to cleanup user", + zap.String("user_id", user.ID), + zap.Error(err), + ) continue } deletedCount++ } - log.Printf("[AccountCleanup] cleaned up %d expired accounts", deletedCount) + zap.L().Info("Cleaned up expired accounts", + zap.Int64("count", deletedCount), + ) return deletedCount, nil } diff --git a/internal/service/call_service.go b/internal/service/call_service.go index 59141a1..f3d1866 100644 --- a/internal/service/call_service.go +++ b/internal/service/call_service.go @@ -8,6 +8,8 @@ import ( "sync" "time" + redispkg "with_you/internal/pkg/redis" + "with_you/internal/config" apperrors "with_you/internal/errors" "with_you/internal/model" @@ -22,8 +24,27 @@ import ( const ( CallLifetimeMs = 60000 // 通话邀请有效期 60秒 (参考 Matrix) CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒 + + redisCallKeyPrefix = "call:" + redisCallByUserPrefix = "call_by_user:" + redisCallAcceptPrefix = "call:accept:" + redisCallTTL = 120 * time.Second + redisCallAcceptLockTTL = 10 * time.Second ) +// activeCallRedisData Redis 中存储的通话数据（精简字段） +type activeCallRedisData struct { + ID string `json:"id"` + ConversationID string `json:"conversation_id"` + CallerID string `json:"caller_id"` + CalleeID string `json:"callee_id"` + CallType string `json:"call_type"` + Status string `json:"status"` + MediaType string `json:"media_type"` + CreatedAt int64 `json:"created_at"` + StartedAt int64 `json:"started_at"` // 0 means nil +} + // ActiveCall 内存中的活跃通话 type ActiveCall struct { ID string @@ -65,9 +86,10 @@ type CallService interface { type callService struct { callRepo repository.CallRepository - hub *ws.Hub + hub ws.MessagePublisher config *config.Config db *gorm.DB + redis *redispkg.Client // 内存中的活跃通话状态 activeCalls map[string]*ActiveCall // callID -> ActiveCall @@ -79,22 +101,24 @@ type callService struct { // NewCallService 创建通话服务 func NewCallService( callRepo repository.CallRepository, - hub *ws.Hub, + publisher ws.MessagePublisher, cfg *config.Config, db *gorm.DB, + redisClient *redispkg.Client, ) CallService { svc := &callService{ callRepo: callRepo, - hub: hub, + hub: publisher, config: cfg, db: db, + redis: redisClient, activeCalls: make(map[string]*ActiveCall), activeCallsByUser: make(map[string]map[string]bool), activeCallsByConv: make(map[string]string), } // 订阅 WebSocket 断开事件 - hub.OnDisconnect(svc.handleUserDisconnect) + publisher.OnDisconnect(svc.handleUserDisconnect) // 自动启动超时清理 svc.StartCleanupTicker() @@ -108,11 +132,159 @@ func (s *callService) generateCallID() string { return strconv.FormatInt(id, 10) } -// getActiveCall 从内存获取活跃通话 +// redisStoreCall 存储通话数据到 Redis +func (s *callService) redisStoreCall(call *ActiveCall) { + if s.redis == nil { + return + } + ctx := context.Background() + + var startedAt int64 + if call.StartedAt != nil { + startedAt = call.StartedAt.UnixMilli() + } + data := activeCallRedisData{ + ID: call.ID, + ConversationID: call.ConversationID, + CallerID: call.CallerID, + CalleeID: call.CalleeID, + CallType: string(call.CallType), + Status: string(call.Status), + MediaType: call.MediaType, + CreatedAt: call.CreatedAt.UnixMilli(), + StartedAt: startedAt, + } + jsonData, err := json.Marshal(data) + if err != nil { + zap.L().Error("Failed to marshal call for Redis", zap.Error(err)) + return + } + + callKey := redisCallKeyPrefix + call.ID + callerKey := redisCallByUserPrefix + call.CallerID + calleeKey := redisCallByUserPrefix + call.CalleeID + + pipe := s.redis.Pipeline() + pipe.Set(ctx, callKey, jsonData, redisCallTTL) + pipe.Set(ctx, callerKey, call.ID, redisCallTTL) + pipe.Set(ctx, calleeKey, call.ID, redisCallTTL) + if _, err := pipe.Exec(ctx); err != nil { + zap.L().Error("Failed to store call in Redis", zap.String("call_id", call.ID), zap.Error(err)) + } +} + +// redisRemoveCall 从 Redis 移除通话数据 +func (s *callService) redisRemoveCall(call *ActiveCall) { + if s.redis == nil { + return + } + ctx := context.Background() + + keys := []string{ + redisCallKeyPrefix + call.ID, + redisCallByUserPrefix + call.CallerID, + redisCallByUserPrefix + call.CalleeID, + redisCallAcceptPrefix + call.ID, + } + if err := s.redis.Del(ctx, keys...); err != nil { + zap.L().Error("Failed to remove call from Redis", zap.String("call_id", call.ID), zap.Error(err)) + } +} + +// redisGetCall 从 Redis 获取通话数据 +func (s *callService) redisGetCall(callID string) *ActiveCall { + if s.redis == nil { + return nil + } + ctx := context.Background() + + raw, err := s.redis.Get(ctx, redisCallKeyPrefix+callID) + if err != nil { + return nil + } + + var data activeCallRedisData + if err := json.Unmarshal([]byte(raw), &data); err != nil { + zap.L().Error("Failed to unmarshal call from Redis", zap.Error(err)) + return nil + } + + call := &ActiveCall{ + ID: data.ID, + ConversationID: data.ConversationID, + CallerID: data.CallerID, + CalleeID: data.CalleeID, + CallType: model.CallType(data.CallType), + Status: model.CallStatus(data.Status), + MediaType: data.MediaType, + ICEServers: s.config.WebRTC.ICEServers, + Participants: map[string]*ActiveParticipant{ + data.CallerID: {UserID: data.CallerID, Status: model.ParticipantStatusJoined}, + data.CalleeID: {UserID: data.CalleeID, Status: model.ParticipantStatusInvited}, + }, + } + + call.CreatedAt = time.UnixMilli(data.CreatedAt) + if data.StartedAt > 0 { + t := time.UnixMilli(data.StartedAt) + call.StartedAt = &t + // 如果已经开始，被叫方也已加入 + call.Participants[data.CalleeID] = &ActiveParticipant{ + UserID: data.CalleeID, + Status: model.ParticipantStatusJoined, + JoinedAt: call.StartedAt, + } + } + + return call +} + +// redisRefreshTTL 刷新通话相关 Redis 键的 TTL +func (s *callService) redisRefreshTTL(call *ActiveCall) { + if s.redis == nil { + return + } + ctx := context.Background() + + keys := []string{ + redisCallKeyPrefix + call.ID, + redisCallByUserPrefix + call.CallerID, + redisCallByUserPrefix + call.CalleeID, + } + for _, key := range keys { + _, _ = s.redis.Expire(ctx, key, redisCallTTL) + } +} + +// redisTryAcceptLock 尝试获取 Accept 分布式锁 +func (s *callService) redisTryAcceptLock(callID string) bool { + if s.redis == nil { + return true // 无 Redis 时允许通过（降级为本地锁） + } + ctx := context.Background() + ok, err := s.redis.GetClient().SetNX(ctx, redisCallAcceptPrefix+callID, "1", redisCallAcceptLockTTL).Result() + if err != nil { + zap.L().Error("Redis SETNX error, allowing accept", zap.Error(err)) + return true + } + return ok +} + +// getActiveCall 从内存获取活跃通话，L1 未命中时回退到 Redis func (s *callService) getActiveCall(callID string) *ActiveCall { s.mu.RLock() - defer s.mu.RUnlock() - return s.activeCalls[callID] + call := s.activeCalls[callID] + s.mu.RUnlock() + if call != nil { + return call + } + + // L1 miss → Redis fallback + call = s.redisGetCall(callID) + if call != nil { + s.storeActiveCall(call) + } + return call } // getActiveCallByConversation 从内存获取会话的活跃通话 @@ -224,6 +396,9 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati // 存储到内存 s.storeActiveCall(call) + // 写入 Redis + s.redisStoreCall(call) + // 检查被叫方是否在线 calleeOnline := s.hub.HasClients(calleeID) @@ -272,6 +447,11 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*Activ return nil, apperrors.ErrCallAlreadyAnswered } + // 分布式锁：防止跨实例重复 Accept + if !s.redisTryAcceptLock(callID) { + return nil, apperrors.ErrCallAlreadyAnswered + } + // 使用原子操作更新状态 s.mu.Lock() if call.Status != model.CallStatusCalling { @@ -285,6 +465,10 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*Activ call.Participants[userID].JoinedAt = &now s.mu.Unlock() + // 更新 Redis + s.redisStoreCall(call) + s.redisRefreshTTL(call) + // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_accepted", map[string]any{ "call_id": callID, @@ -324,6 +508,9 @@ func (s *callService) Reject(ctx context.Context, callID, userID string) error { // 从内存移除 s.removeActiveCall(callID) + // 从 Redis 移除 + s.redisRemoveCall(call) + // 保存到数据库作为历史记录 s.saveCallHistory(call, model.CallStatusRejected, 0) @@ -358,6 +545,9 @@ func (s *callService) Busy(ctx context.Context, callID, userID string) error { // 从内存移除 s.removeActiveCall(callID) + // 从 Redis 移除 + s.redisRemoveCall(call) + // 保存到数据库作为历史记录 s.saveCallHistory(call, model.CallStatusMissed, 0) @@ -414,6 +604,9 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str // 从内存移除 s.removeActiveCall(callID) + // 从 Redis 移除 + s.redisRemoveCall(call) + // 保存到数据库作为历史记录 s.saveCallHistory(call, endStatus, duration) diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index dede620..460bbde 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -67,7 +67,7 @@ type chatServiceImpl struct { repo repository.MessageRepository userRepo repository.UserRepository sensitive SensitiveService - wsHub *ws.Hub + wsHub ws.MessagePublisher pushSvc PushService cache cache.Cache @@ -80,7 +80,7 @@ func NewChatService( repo repository.MessageRepository, userRepo repository.UserRepository, sensitive SensitiveService, - wsHub *ws.Hub, + publisher ws.MessagePublisher, cacheBackend cache.Cache, uploadService *UploadService, pushSvc PushService, @@ -101,7 +101,7 @@ func NewChatService( repo: repo, userRepo: userRepo, sensitive: sensitive, - wsHub: wsHub, + wsHub: publisher, pushSvc: pushSvc, cache: cacheBackend, conversationCache: conversationCache, diff --git a/internal/service/group_service.go b/internal/service/group_service.go index 9113989..87565d3 100644 --- a/internal/service/group_service.go +++ b/internal/service/group_service.go @@ -112,13 +112,13 @@ type groupService struct { messageRepo repository.MessageRepository requestRepo repository.GroupJoinRequestRepository notifyRepo repository.SystemNotificationRepository - wsHub *ws.Hub + wsHub ws.MessagePublisher cache cache.Cache conversationCache *cache.ConversationCache } // NewGroupService 创建群组服务 -func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.UserRepository, messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, wsHub *ws.Hub, cacheBackend cache.Cache) GroupService { +func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.UserRepository, messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, publisher ws.MessagePublisher, cacheBackend cache.Cache) GroupService { convRepoAdapter := cache.NewConversationRepositoryAdapter(messageRepo) msgRepoAdapter := cache.NewMessageRepositoryAdapter(messageRepo) conversationCache := cache.NewConversationCache( @@ -134,7 +134,7 @@ func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.U messageRepo: messageRepo, requestRepo: requestRepo, notifyRepo: notifyRepo, - wsHub: wsHub, + wsHub: publisher, cache: cacheBackend, conversationCache: conversationCache, } diff --git a/internal/service/push_service.go b/internal/service/push_service.go index 7ef1246..008c6f6 100644 --- a/internal/service/push_service.go +++ b/internal/service/push_service.go @@ -82,7 +82,7 @@ type pushServiceImpl struct { pushRepo repository.PushRecordRepository deviceRepo repository.DeviceTokenRepository messageRepo repository.MessageRepository - wsHub *ws.Hub + wsHub ws.MessagePublisher jpushClient *jpush.Client pushQueue chan *pushTask @@ -102,14 +102,14 @@ func NewPushService( pushRepo repository.PushRecordRepository, deviceRepo repository.DeviceTokenRepository, messageRepo repository.MessageRepository, - wsHub *ws.Hub, + publisher ws.MessagePublisher, jpushClient *jpush.Client, ) PushService { return &pushServiceImpl{ pushRepo: pushRepo, deviceRepo: deviceRepo, messageRepo: messageRepo, - wsHub: wsHub, + wsHub: publisher, jpushClient: jpushClient, pushQueue: make(chan *pushTask, PushQueueSize), stopChan: make(chan struct{}), diff --git a/internal/service/qrcode_login_service.go b/internal/service/qrcode_login_service.go index 93e347b..e3eb189 100644 --- a/internal/service/qrcode_login_service.go +++ b/internal/service/qrcode_login_service.go @@ -54,7 +54,7 @@ type QRCodeSession struct { // QRCodeLoginService 二维码登录服务 type QRCodeLoginService struct { redis *redis.Client - wsHub *ws.Hub + wsHub ws.MessagePublisher jwtService *JWTService userService UserService activityService UserActivityService @@ -62,10 +62,10 @@ type QRCodeLoginService struct { } // NewQRCodeLoginService 创建二维码登录服务 -func NewQRCodeLoginService(redis *redis.Client, wsHub *ws.Hub, jwtService *JWTService, userService UserService, activityService UserActivityService, logService *LogService) *QRCodeLoginService { +func NewQRCodeLoginService(redis *redis.Client, publisher ws.MessagePublisher, jwtService *JWTService, userService UserService, activityService UserActivityService, logService *LogService) *QRCodeLoginService { return &QRCodeLoginService{ redis: redis, - wsHub: wsHub, + wsHub: publisher, jwtService: jwtService, userService: userService, activityService: activityService, @@ -280,9 +280,16 @@ func (s *QRCodeLoginService) Cancel(ctx context.Context, sessionID, userID strin return nil } -// GetWSHub 获取WS Hub +// GetWSHub 获取底层 WS Hub（用于 Subscribe 等本地操作） func (s *QRCodeLoginService) GetWSHub() *ws.Hub { - return s.wsHub + switch p := s.wsHub.(type) { + case *ws.Bus: + return p.Hub() + case *ws.Hub: + return p + default: + return nil + } } // GetUserService 获取用户服务 diff --git a/internal/service/trade_service.go b/internal/service/trade_service.go index da2ada2..fea0f2f 100644 --- a/internal/service/trade_service.go +++ b/internal/service/trade_service.go @@ -79,8 +79,6 @@ func (s *tradeService) GetByID(ctx context.Context, id string, currentUserID *st } } - _ = s.tradeRepo.IncrementViews(ctx, id) - return dto.ConvertTradeItemToResponse(item, isFavorited), nil } diff --git a/internal/service/user_profile_audit_service.go b/internal/service/user_profile_audit_service.go index 3e7bbcc..77590a1 100644 --- a/internal/service/user_profile_audit_service.go +++ b/internal/service/user_profile_audit_service.go @@ -2,7 +2,6 @@ package service import ( "context" - "log" "time" "with_you/internal/model" @@ -179,7 +178,10 @@ func (s *userProfileAuditServiceImpl) GetPendingCounts(ctx context.Context) (int func (s *userProfileAuditServiceImpl) reviewAvatarAsync(auditID, userID, avatarURL string) { defer func() { if r := recover(); r != nil { - log.Printf("[ERROR] Panic in avatar moderation async flow, audit=%s panic=%v", auditID, r) + zap.L().Error("Panic in avatar moderation async flow", + zap.String("audit_id", auditID), + zap.Any("panic", r), + ) } }() @@ -245,7 +247,10 @@ func (s *userProfileAuditServiceImpl) reviewAvatarAsync(auditID, userID, avatarU func (s *userProfileAuditServiceImpl) reviewCoverAsync(auditID, userID, coverURL string) { defer func() { if r := recover(); r != nil { - log.Printf("[ERROR] Panic in cover moderation async flow, audit=%s panic=%v", auditID, r) + zap.L().Error("Panic in cover moderation async flow", + zap.String("audit_id", auditID), + zap.Any("panic", r), + ) } }() @@ -311,7 +316,10 @@ func (s *userProfileAuditServiceImpl) reviewCoverAsync(auditID, userID, coverURL func (s *userProfileAuditServiceImpl) reviewBioAsync(auditID, userID, bio string) { defer func() { if r := recover(); r != nil { - log.Printf("[ERROR] Panic in bio moderation async flow, audit=%s panic=%v", auditID, r) + zap.L().Error("Panic in bio moderation async flow", + zap.String("audit_id", auditID), + zap.Any("panic", r), + ) } }() diff --git a/internal/wire/handler.go b/internal/wire/handler.go index cf67305..87815c2 100644 --- a/internal/wire/handler.go +++ b/internal/wire/handler.go @@ -63,9 +63,9 @@ func ProvideMessageHandler( messageService *service.MessageService, userService service.UserService, groupService service.GroupService, - wsHub *ws.Hub, + publisher ws.MessagePublisher, ) *handler.MessageHandler { - return handler.NewMessageHandler(chatService, messageService, userService, groupService, wsHub) + return handler.NewMessageHandler(chatService, messageService, userService, groupService, publisher) } // ProvideSystemMessageHandler 提供系统消息处理器 @@ -93,14 +93,14 @@ func ProvideScheduleHandler( // ProvideWSHandler 提供WebSocket处理器 func ProvideWSHandler( - wsHub *ws.Hub, + publisher ws.MessagePublisher, chatService service.ChatService, groupService service.GroupService, jwtService *service.JWTService, callService service.CallService, userRepo repository.UserRepository, ) *handler.WSHandler { - return handler.NewWSHandler(wsHub, chatService, groupService, jwtService, callService, userRepo) + return handler.NewWSHandler(publisher, chatService, groupService, jwtService, callService, userRepo) } // ProvideAdminVerificationHandler 提供管理端身份认证处理器 diff --git a/internal/wire/infrastructure.go b/internal/wire/infrastructure.go index 85afe30..862ac4f 100644 --- a/internal/wire/infrastructure.go +++ b/internal/wire/infrastructure.go @@ -46,8 +46,8 @@ var InfrastructureSet = wire.NewSet( ProvideBuiltinHooks, ProvideModerationHooks, - // WebSocket Hub - ProvideWSHub, + // WebSocket MessagePublisher + ProvideWSMessagePublisher, // 外部服务客户端 ProvideOpenAIClient, @@ -163,9 +163,32 @@ func ProvideModerationHooks( return moderationHooks } -// ProvideWSHub 提供 WebSocket Hub -func ProvideWSHub() *ws.Hub { - return ws.NewHub() +// ProvideWSMessagePublisher 提供 WebSocket 消息发布器 +// standalone 模式返回 *Hub，cluster 模式返回 *Bus（组合 Hub + Redis Pub/Sub） +func ProvideWSMessagePublisher(cfg *config.Config, redisClient *redis.Client) ws.MessagePublisher { + hub := ws.NewHub() + + if cfg.WebSocket.Mode == "cluster" && redisClient != nil { + bus := ws.NewBus(hub, redisClient.GetClient(), &ws.WSClusterConfig{ + InstanceID: cfg.WebSocket.Cluster.InstanceID, + MsgChannel: cfg.WebSocket.Cluster.MsgChannel, + OnlineTTL: cfg.WebSocket.Cluster.OnlineTTL, + HeartbeatInterval: cfg.WebSocket.Cluster.HeartbeatInterval, + }) + if err := bus.Start(); err != nil { + zap.L().Error("Failed to start WebSocket Bus, falling back to standalone mode", + zap.Error(err), + ) + return hub + } + zap.L().Info("WebSocket running in cluster mode", + zap.String("instance_id", bus.InstanceID()), + ) + return bus + } + + zap.L().Info("WebSocket running in standalone mode") + return hub } // ProvideOpenAIClient 提供 OpenAI 客户端 diff --git a/internal/wire/service.go b/internal/wire/service.go index 7462551..8779e7b 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -105,10 +105,10 @@ func ProvidePushService( pushRepo repository.PushRecordRepository, deviceTokenRepo repository.DeviceTokenRepository, messageRepo repository.MessageRepository, - wsHub *ws.Hub, + publisher ws.MessagePublisher, jpushClient *jpush.Client, ) service.PushService { - return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, wsHub, jpushClient) + return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, publisher, jpushClient) } // ProvideSystemMessageService 提供系统消息服务 @@ -205,12 +205,12 @@ func ProvideChannelService(channelRepo repository.ChannelRepository, cacheBacken func ProvideChatService( messageRepo repository.MessageRepository, userRepo repository.UserRepository, - wsHub *ws.Hub, + publisher ws.MessagePublisher, cacheBackend cache.Cache, uploadService *service.UploadService, pushService service.PushService, ) service.ChatService { - return service.NewChatService(messageRepo, userRepo, nil, wsHub, cacheBackend, uploadService, pushService) + return service.NewChatService(messageRepo, userRepo, nil, publisher, cacheBackend, uploadService, pushService) } // ProvideScheduleService 提供日程服务 @@ -236,10 +236,10 @@ func ProvideGroupService( messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, - wsHub *ws.Hub, + publisher ws.MessagePublisher, cacheBackend cache.Cache, ) service.GroupService { - return service.NewGroupService(groupRepo, userRepo, messageRepo, requestRepo, notifyRepo, wsHub, cacheBackend) + return service.NewGroupService(groupRepo, userRepo, messageRepo, requestRepo, notifyRepo, publisher, cacheBackend) } // ProvideUploadService 提供上传服务 @@ -382,13 +382,13 @@ func ProvideHotRankWorker(cfg *config.Config, postRepo repository.PostRepository func ProvideQRCodeLoginService( redisClient *redis.Client, - wsHub *ws.Hub, + publisher ws.MessagePublisher, jwtService *service.JWTService, userService service.UserService, activityService service.UserActivityService, logService *service.LogService, ) *service.QRCodeLoginService { - return service.NewQRCodeLoginService(redisClient, wsHub, jwtService, userService, activityService, logService) + return service.NewQRCodeLoginService(redisClient, publisher, jwtService, userService, activityService, logService) } // ProvideMaterialService 提供学习资料服务 @@ -407,11 +407,12 @@ func parseDuration(s string) (time.Duration, error) { // ProvideCallService 提供通话服务 func ProvideCallService( callRepo repository.CallRepository, - wsHub *ws.Hub, + publisher ws.MessagePublisher, cfg *config.Config, db *gorm.DB, + redisClient *redis.Client, ) service.CallService { - return service.NewCallService(callRepo, wsHub, cfg, db) + return service.NewCallService(callRepo, publisher, cfg, db, redisClient) } // ProvideReportService 提供举报服务