diff --git a/backend/gateway/docs/docs.go b/backend/gateway/docs/docs.go index 5bf1d95..8b288b1 100644 --- a/backend/gateway/docs/docs.go +++ b/backend/gateway/docs/docs.go @@ -5130,6 +5130,10 @@ const docTemplate = `{ "current_identity": { "$ref": "#/definitions/dto.CurrentIdentityDTO" }, + "mobile": { + "description": "脱敏手机号，如 139****0001", + "type": "string" + }, "nickname": { "type": "string" }, diff --git a/backend/gateway/dto/response_dto.go b/backend/gateway/dto/response_dto.go index a0126ec..094beec 100644 --- a/backend/gateway/dto/response_dto.go +++ b/backend/gateway/dto/response_dto.go @@ -60,6 +60,7 @@ type GetMeResponseDTO struct { Nickname string `json:"nickname"` AvatarURL string `json:"avatar_url,omitempty"` ChainAddress string `json:"chain_address,omitempty"` + Mobile string `json:"mobile,omitempty"` // 脱敏手机号，如 139****0001 CurrentIdentity CurrentIdentityDTO `json:"current_identity"` } diff --git a/backend/gateway/dto/user_converter.go b/backend/gateway/dto/user_converter.go index 16c39c0..0289de5 100644 --- a/backend/gateway/dto/user_converter.go +++ b/backend/gateway/dto/user_converter.go @@ -118,6 +118,11 @@ func ToGetMeResponseDTO(user *pb.User, profile *pb.FanProfile, star *pb.Star) Ge dto.ChainAddress = profile.ChainAddress } + // 脱敏手机号：139****0001 + if user.Mobile != "" && len(user.Mobile) == 11 { + dto.Mobile = user.Mobile[:3] + "****" + user.Mobile[7:] + } + return dto } diff --git a/backend/gateway/main.go b/backend/gateway/main.go index 034c4d0..5dcacbb 100644 --- a/backend/gateway/main.go +++ b/backend/gateway/main.go @@ -225,8 +225,12 @@ func main() { // 4.13 初始化 Activity Hub（WebSocket 实时推送） redisClient := database.GetRedis() activityHub := socket.NewActivityHub(redisClient, cfg.WebSocket.ActivityPath) - go activityHub.Run(context.Background()) - defer activityHub.Close() + hubCtx, hubCancel := context.WithCancel(context.Background()) + go activityHub.Run(hubCtx) + defer func() { + hubCancel() + activityHub.Close() + }() logger.Logger.Info("ActivityHub initialized", zap.String("path", cfg.WebSocket.ActivityPath), zap.Bool("redis_available", redisClient != nil), @@ -261,5 +265,7 @@ func main() { <-quit logger.Logger.Info("Shutting down gateway server...") + hubCancel() + activityHub.Close() logger.Logger.Info("Gateway server stopped") } diff --git a/backend/gateway/socket/activity_socket.go b/backend/gateway/socket/activity_socket.go index 3d6e6e7..a04e3b5 100644 --- a/backend/gateway/socket/activity_socket.go +++ b/backend/gateway/socket/activity_socket.go @@ -58,13 +58,41 @@ func (h *ActivityHub) ActivityPath() string { return h.activityPath } -// Run 启动 Redis PSubscribe，收到 publish 后 fanout 到本地连接 +// Run 启动 Redis PSubscribe 主循环，带自动重连。 +// 当 Redis 不可用时定时重试，不会永久阻塞； +// ch 意外关闭后自动重新订阅，确保 fanout 链路始终存活。 func (h *ActivityHub) Run(ctx context.Context) { + for { + select { + case <-ctx.Done(): + logger.Logger.Info("ActivityHub Run loop exiting due to context done") + return + default: + } + + h.runOnce(ctx) + + // runOnce 退出说明 PSubscribe 断开或 Redis 不可用；等待后重试 + logger.Logger.Warn("ActivityHub runOnce exited, will retry after 5s") + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + } +} + +// runOnce 执行单次 PSubscribe → fanout 循环。 +func (h *ActivityHub) runOnce(ctx context.Context) { if h.redisClient == nil { - logger.Logger.Warn("ActivityHub: redisClient is nil, Pub/Sub fanout disabled") - <-ctx.Done() + logger.Logger.Warn("ActivityHub: redisClient is nil, Pub/Sub fanout disabled, waiting...") + select { + case <-ctx.Done(): + case <-time.After(5 * time.Second): + } return } + sub := h.redisClient.PSubscribe(ctx, "act:*:messages", "act:*:contributions") defer sub.Close() ch := sub.Channel() @@ -72,11 +100,11 @@ func (h *ActivityHub) Run(ctx context.Context) { for { select { case <-ctx.Done(): - logger.Logger.Info("ActivityHub Run loop exiting due to context done") + logger.Logger.Info("ActivityHub runOnce exiting due to context done") return case msg, ok := <-ch: if !ok { - logger.Logger.Warn("ActivityHub Redis Pub/Sub channel closed") + logger.Logger.Warn("ActivityHub Redis Pub/Sub channel closed, will reconnect") return } var payload map[string]interface{} @@ -89,7 +117,8 @@ func (h *ActivityHub) Run(ctx context.Context) { } } -// fanout 把 payload 推送到订阅该 channel 的所有本地连接 +// fanout 把 payload 推送到订阅该 channel 的所有本地连接。 +// 写失败时主动清理死连接，不等 Ping/Pong 超时（最长 ~90s）。 func (h *ActivityHub) fanout(channel string, payload map[string]interface{}) { h.mu.RLock() conns := h.subscriptions[channel] @@ -99,11 +128,22 @@ func (h *ActivityHub) fanout(channel string, payload map[string]interface{}) { } h.mu.RUnlock() + var dead []*ActivityConn for _, c := range targets { if err := c.writeJSON(payload); err != nil { logger.Logger.Error("ActivityHub writeJSON failed", zap.Int64("user_id", c.UserID), zap.Error(err)) + dead = append(dead, c) } } + + // 批量清理死连接（需要写锁，与 RLock 分离以避免死锁） + if len(dead) > 0 { + h.mu.Lock() + for _, c := range dead { + h.removeConnLocked(c) + } + h.mu.Unlock() + } } // HandleWebSocket 处理 /activity 握手 @@ -148,12 +188,17 @@ func (h *ActivityHub) HandleWebSocket(w http.ResponseWriter, r *http.Request) { zap.Int64("star_id", starID), ) - // 立即推送 auth_response + // 立即推送 auth_response，附带 pub/sub 状态 + pubsubStatus := "connected" + if h.redisClient == nil { + pubsubStatus = "pubsub_disabled" + } _ = conn.WriteJSON(map[string]interface{}{ - "type": "auth_response", - "success": true, - "user_id": userID, - "star_id": starID, + "type": "auth_response", + "success": true, + "user_id": userID, + "star_id": starID, + "pubsub_status": pubsubStatus, }) go c.readPump() @@ -309,7 +354,12 @@ func (h *ActivityHub) unsubscribe(c *ActivityConn, activityID int64, topics []st // unregister 断开时清理 func (h *ActivityHub) unregister(c *ActivityConn) { h.mu.Lock() - defer h.mu.Unlock() + h.removeConnLocked(c) + h.mu.Unlock() +} + +// removeConnLocked 从 clients 和所有 subscriptions 中移除连接（需持有 h.mu 写锁） +func (h *ActivityHub) removeConnLocked(c *ActivityConn) { if conns, ok := h.clients[c.UserID]; ok { delete(conns, c) if len(conns) == 0 { diff --git a/backend/gateway/socket/ai_chat_socket.go b/backend/gateway/socket/ai_chat_socket.go index bee3ee4..d032283 100644 --- a/backend/gateway/socket/ai_chat_socket.go +++ b/backend/gateway/socket/ai_chat_socket.go @@ -31,8 +31,8 @@ var upgrader = websocket.Upgrader{ // Hub 管理所有 AI Chat WebSocket 连接 type Hub struct { - // 用户连接映射: userId -> *Connection - clients map[int64]*Connection + // 用户连接映射: userId -> set of *Connection（支持同一用户多设备） + clients map[int64]map[*Connection]struct{} // Dubbo 客户端 aiChatClient *client.Client @@ -69,7 +69,7 @@ func (c *Connection) sendError(code, message string) { // NewHub 创建 Hub 实例 func NewHub(aiChatClient *client.Client, aiChatPath string) *Hub { return &Hub{ - clients: make(map[int64]*Connection), + clients: make(map[int64]map[*Connection]struct{}), aiChatClient: aiChatClient, aiChatPath: aiChatPath, } @@ -113,9 +113,12 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) { Hub: h, } - // 注册连接 + // 注册连接（同一用户多设备各自独立连接） h.mu.Lock() - h.clients[userID] = connection + if h.clients[userID] == nil { + h.clients[userID] = make(map[*Connection]struct{}) + } + h.clients[userID][connection] = struct{}{} h.mu.Unlock() logger.Logger.Info("WebSocket connection established", @@ -166,7 +169,12 @@ func (h *Hub) validateToken(token string) (int64, int64, error) { func (c *Connection) readPump() { defer func() { c.Hub.mu.Lock() - delete(c.Hub.clients, c.UserID) + if conns, ok := c.Hub.clients[c.UserID]; ok { + delete(conns, c) + if len(conns) == 0 { + delete(c.Hub.clients, c.UserID) + } + } c.Hub.mu.Unlock() c.Conn.Close() }() @@ -467,12 +475,14 @@ func (c *Connection) getPersonas() { // sendError 发送错误消息 (使用 Send 通道) -// Close 关闭连接 +// Close 关闭所有连接 func (h *Hub) Close() { h.mu.Lock() defer h.mu.Unlock() - for _, conn := range h.clients { - conn.Conn.Close() + for _, conns := range h.clients { + for c := range conns { + c.Conn.Close() + } } } \ No newline at end of file diff --git a/backend/services/userService/service/auth_service.go b/backend/services/userService/service/auth_service.go index 81673d9..40d7089 100644 --- a/backend/services/userService/service/auth_service.go +++ b/backend/services/userService/service/auth_service.go @@ -178,6 +178,7 @@ func (s *authService) Register(ctx context.Context, req *pb.RegisterRequest) (*p UserID: user.ID, StarID: req.StarId, Nickname: req.Nickname, + AvatarURL: user.AvatarURL, Level: 1, Times: 1, Social: 0, diff --git a/backend/services/userService/service/identity_service.go b/backend/services/userService/service/identity_service.go index 1137d48..64fe51a 100644 --- a/backend/services/userService/service/identity_service.go +++ b/backend/services/userService/service/identity_service.go @@ -258,12 +258,20 @@ func (s *identityService) AddIdentity(req *pb.AddIdentityRequest, userID int64) return nil, fmt.Errorf("failed to check existing fan profile: %w", err) } - // 5. 创建新的粉丝档案 + // 5. 查询用户头像（用于新粉丝档案） + var avatarURL *string + user, err := s.userRepo.GetByID(userID) + if err == nil && user != nil { + avatarURL = user.AvatarURL + } + + // 6. 创建新的粉丝档案 now := time.Now().UnixMilli() fanProfile := &models.FanProfile{ UserID: userID, StarID: req.StarId, Nickname: req.Nickname, + AvatarURL: avatarURL, Level: 1, Times: 1, Social: 0, diff --git a/frontend/App.vue b/frontend/App.vue index 38eb576..0e2537a 100644 --- a/frontend/App.vue +++ b/frontend/App.vue @@ -21,8 +21,19 @@ export default { // 冷启动时检查本地 token，标记是否需要跳转到一键登录页 // 仅 onLaunch 触发（冷启动/app 销毁后重新打开），后台切回前台不触发 + // 通过 HIDE_TIME_KEY 判断 App 是否短时间内被系统重启（卡顿/网络切换/内存回收）： + // 若切后台 5 分钟内又触发 onLaunch，说明是系统被动重启，跳过 quickLogin 保持原页面 const token = uni.getStorageSync("access_token"); - this.globalData.shouldShowQuickLogin = !!token; + const lastHideTime = uni.getStorageSync(HIDE_TIME_KEY) || 0; + const RECENT_RESTART_THRESHOLD = 5 * 60 * 1000; // 5 分钟 + const isRecentRestart = Date.now() - lastHideTime < RECENT_RESTART_THRESHOLD; + this.globalData.shouldShowQuickLogin = !!token && !isRecentRestart; + + // 系统被动重启时，清除 needs_welcome，防止回到 square 时误触发 TopfansWelcome + // （welcomeShownThisSession 会被重置为 false，若 needs_welcome 还在 storage 就会弹欢迎页） + if (isRecentRestart) { + uni.removeStorageSync("needs_welcome"); + } // 【TopfansWelcome 会话判定】 // 触发场景（即用户期望展示欢迎页的入口）： // 1) "后台关闭才会在打开"：用户在后台杀掉进程后重新打开 → onLaunch 触发 → 新会话 @@ -46,6 +57,9 @@ export default { // 后台切回前台时 shouldShowQuickLogin 为 false，不跳转，直接回到挂载时的页面 if (this.globalData.shouldShowQuickLogin) { this.globalData.shouldShowQuickLogin = false; + // 清除 needs_welcome，避免 App 重启后 quickLogin → square 误触发 TopfansWelcome + // 首次登录/注册的欢迎页由 login.vue / selectRole.vue 各自负责写入 + uni.removeStorageSync("needs_welcome"); // 即使跳转到一键登录，仍需初始化推送服务和清除角标 this.getAllNotice(); this.clearBadgeAndNotifications(); @@ -55,6 +69,9 @@ export default { this.handleBackgroundReturn(); + // 从后台切回前台时重新初始化 WebSocket（onHide 中已关闭） + this.initWebSocket(); + this.getAllNotice(); // 打开 App 时清除图标角标和通知栏 diff --git a/frontend/pages/ai-dazi/index.vue b/frontend/pages/ai-dazi/index.vue index 56c9530..20e9243 100644 --- a/frontend/pages/ai-dazi/index.vue +++ b/frontend/pages/ai-dazi/index.vue @@ -172,8 +172,8 @@ export default { } this.isErrorProcessing = true this.isTyping = false - uni.showToast({ title: data.message || data.error || '发生错误', icon: 'none' }) this.aiMessage = '亲爱的你来辣 ~~' + uni.showToast({ title: data.message || data.error || '发生错误', icon: 'none' }) // 延迟重置标志 setTimeout(() => { this.isErrorProcessing = false diff --git a/frontend/pages/login/quickLogin.vue b/frontend/pages/login/quickLogin.vue index 3859048..af17382 100644 --- a/frontend/pages/login/quickLogin.vue +++ b/frontend/pages/login/quickLogin.vue @@ -132,7 +132,7 @@ const maskPhone = (phone) => { }; const maskedPhone = ref( - userInfo.mobile_masked || maskPhone(userInfo.mobile) || "174****2223", + userInfo.mobile_masked|| userInfo.mobile || maskPhone(userInfo.mobile) || "174****2223", ); const userAvatar = ref( userInfo.avatar_url || "/static/login/portal/quicklogin-avatar.png", @@ -157,8 +157,11 @@ const handleQuickLogin = () => { uni.showLoading({ title: "登录中..." }); setTimeout(() => { uni.hideLoading(); - // 标记从一键登录进入，触发 TopfansWelcome 显示 - uni.setStorageSync("needs_welcome", true); + // 标记从一键登录进入，触发 TopfansWelcome 显示 + uni.setStorageSync("needs_welcome", true); + // 原因：quickLogin 只在 App 冷启动 + 有 token 时由 App.vue 跳转进入， + // 若写入 needs_welcome 则每次系统杀掉进程重启都会触发 TopfansWelcome， + // 用户体验很差。needs_welcome 由 login.vue（密码登录）和 selectRole.vue（注册选角）负责写入。 uni.reLaunch({ url: "/pages/square/square" }); }, 800); }; diff --git a/frontend/pages/profile/selectRole.vue b/frontend/pages/profile/selectRole.vue index d4d1bbd..a92a137 100644 --- a/frontend/pages/profile/selectRole.vue +++ b/frontend/pages/profile/selectRole.vue @@ -170,7 +170,7 @@ const selectStar = (star) => { // 重新选择 const resetSelection = () => { selectedStar.value = null; - searchText.value = ""; + // searchText.value = ""; }; // 确认选择明星（从输入框内容自动匹配） diff --git a/frontend/pages/support-activity/composables/useContributionRealtime.js b/frontend/pages/support-activity/composables/useContributionRealtime.js index cd1f05a..6dec82b 100644 --- a/frontend/pages/support-activity/composables/useContributionRealtime.js +++ b/frontend/pages/support-activity/composables/useContributionRealtime.js @@ -70,6 +70,11 @@ export function useContributionRealtime(activityId, isPageActive) { function onWsConnect() { if (usingWS) return + // 即使 WS 已连接，如果后端 Pub/Sub 不可用，也不切到 WS，保持轮询 + if (!socket.isPubSubEnabled()) { + console.log('[useContributionRealtime] WS connected but pubsub disabled, keeping polling') + return + } usingWS = true stopPolling() // 停掉可能的轮询 socket.subscribe(activityId.value, ['contributions']) @@ -81,9 +86,19 @@ export function useContributionRealtime(activityId, isPageActive) { startPolling() // 降级为轮询 } + // 后端上报 pub/sub 不可用：如果之前已切到 WS，回退到轮询 + function onPubsubDisabled() { + if (!usingWS) return + console.warn('[useContributionRealtime] Pub/Sub disabled, falling back to polling') + usingWS = false + socket.unsubscribe(activityId.value, ['contributions']) + startPolling() + } + socket.onContributionsResponse(onWsMessage) socket.on('connect', onWsConnect) socket.on('disconnect', onWsDisconnect) + socket.on('pubsub_disabled', onPubsubDisabled) onMounted(() => { // 总是调用 connect():SocketManager.connect() 内部会判断 token 是否变化 @@ -93,11 +108,12 @@ export function useContributionRealtime(activityId, isPageActive) { if (token) { socket.connect(token).catch(err => console.warn('[useContributionRealtime] connect error:', err)) } - // 同步分支:如果 WS 已连接(单例复用导致 'connect' 事件不会再次触发), + // 同步分支:如果 WS 已连接（单例复用导致 'connect' 事件不会再次触发）且 pubsub 可用, // 必须直接调 onWsConnect 停轮询,否则 polling 会一直跑。 // 异步分支:WS 还没连上时,先起 polling 兜底; // 等 'connect' 事件触发 onWsConnect 后会停掉轮询。 - if (socket.isConnected) { + // 但如果 pubsub 被禁用,即使 WS 已连接也不切,保持轮询。 + if (socket.isConnected && socket.isPubSubEnabled()) { onWsConnect() } else { startPolling() @@ -108,6 +124,7 @@ export function useContributionRealtime(activityId, isPageActive) { if (usingWS) socket.unsubscribe(activityId.value, ['contributions']) socket.off('connect', onWsConnect) socket.off('disconnect', onWsDisconnect) + socket.off('pubsub_disabled', onPubsubDisabled) socket.offContributionsResponse(onWsMessage) stopPolling() resetPolling() diff --git a/frontend/pages/support-activity/composables/useMessageRealtime.js b/frontend/pages/support-activity/composables/useMessageRealtime.js index 2e556ad..e9a0d42 100644 --- a/frontend/pages/support-activity/composables/useMessageRealtime.js +++ b/frontend/pages/support-activity/composables/useMessageRealtime.js @@ -94,13 +94,27 @@ export function useMessageRealtime(activityId) { if (token) { socket.connect(token).catch(err => console.warn('[useMessageRealtime] connect error:', err)) } + // 同步分支：如果 WS 已连接（单例复用，另一个 composable 可能先连上了）， + // subscribe 会直接发送；否则缓存到 _topics，等待 connect 事件触发 resubscribeAll socket.subscribe(activityId.value, ['messages']) socket.onMessagesResponse(onWsMessage) + + // 如果后端 pub/sub 被禁用，WS 订阅无效，需要手动刷新才能看到新消息 + if (!socket.isPubSubEnabled()) { + console.warn('[useMessageRealtime] Pub/Sub disabled, real-time messages unavailable. Pull-to-refresh to see new messages.') + } }) + // 后端 pub/sub 状态变化：禁用时提醒用户 + function onPubsubDisabled() { + console.warn('[useMessageRealtime] Pub/Sub became disabled, new messages will not arrive in real-time') + } + socket.on('pubsub_disabled', onPubsubDisabled) + onUnmounted(() => { socket.unsubscribe(activityId.value, ['messages']) socket.offMessagesResponse(onWsMessage) + socket.off('pubsub_disabled', onPubsubDisabled) }) return { diff --git a/frontend/store/modules/user.js b/frontend/store/modules/user.js index 4c54664..be4f80e 100644 --- a/frontend/store/modules/user.js +++ b/frontend/store/modules/user.js @@ -145,7 +145,7 @@ const actions = { // 缓存登录手机号 const loginMobile = mobile - uni.setStorageSync('login_mobile', loginMobile) + // uni.setStorageSync('login_mobile', loginMobile) uni.setStorageSync('user', JSON.stringify(user)) commit('SET_USER_INFO', user) diff --git a/frontend/utils/socket/GlobalSocketManager.js b/frontend/utils/socket/GlobalSocketManager.js index 976e8ed..1b6f7c2 100644 --- a/frontend/utils/socket/GlobalSocketManager.js +++ b/frontend/utils/socket/GlobalSocketManager.js @@ -10,13 +10,25 @@ class GlobalSocketManager { this.sockets = {} // serviceName -> SocketManager this.token = null this.isAllConnected = false + this._initialized = false // 防止重复 init 导致监听器累积 + // 保存监听器引用，用于 cleanup + this._aiChatConnectHandler = null + this._aiChatErrorHandler = null + this._activityConnectHandler = null + this._activityErrorHandler = null } /** - * 初始化所有连接 + * 初始化所有连接（幂等，重复调用会先清理旧监听器再重连） */ init(token) { this.token = token + + // 总是先清理旧监听器再注册新的，防止累积。 + // 注意：closeAll() 会清空 this.sockets，所以 cleanup 必须通过单例 getter 获取 socket 引用。 + this._cleanupListeners() + this._initialized = true + this._initAiChat() this._initActivity() // Future: this._initNotification() @@ -24,22 +36,49 @@ class GlobalSocketManager { async _initAiChat() { const aiChat = getAiChatSocket() - aiChat.on('connect', () => console.log('AI Chat connected')) - aiChat.on('error', (err) => console.error('AI Chat error:', err)) + this._aiChatConnectHandler = () => console.log('AI Chat connected') + this._aiChatErrorHandler = (err) => console.error('AI Chat error:', err) + aiChat.on('connect', this._aiChatConnectHandler) + aiChat.on('error', this._aiChatErrorHandler) await aiChat.connect(this.token) this.sockets['ai_chat'] = aiChat } async _initActivity() { const activity = getActivitySocket() - activity.on('connect', () => console.log('Activity socket connected')) - activity.on('error', (err) => console.error('Activity socket error:', err)) + this._activityConnectHandler = () => console.log('Activity socket connected') + this._activityErrorHandler = (err) => console.error('Activity socket error:', err) + activity.on('connect', this._activityConnectHandler) + activity.on('error', this._activityErrorHandler) if (this.token) { await activity.connect(this.token) } this.sockets['activity'] = activity } + /** + * 清理之前注册的监听器，防止每次 init() 累积。 + * 通过单例 getter 获取 socket 引用（而非 this.sockets）， + * 因为 closeAll() 会清空 this.sockets 但单例仍存活。 + */ + _cleanupListeners() { + try { + const aiChat = getAiChatSocket() + if (this._aiChatConnectHandler) { + aiChat.off('connect', this._aiChatConnectHandler) + aiChat.off('error', this._aiChatErrorHandler) + } + } catch (e) { /* singleton not yet created, skip */ } + + try { + const activity = getActivitySocket() + if (this._activityConnectHandler) { + activity.off('connect', this._activityConnectHandler) + activity.off('error', this._activityErrorHandler) + } + } catch (e) { /* singleton not yet created, skip */ } + } + getSocket(serviceName) { return this.sockets[serviceName] } @@ -47,6 +86,7 @@ class GlobalSocketManager { closeAll() { Object.values(this.sockets).forEach(socket => socket.close()) this.sockets = {} + this._initialized = false // 关闭后允许下次 init() 重新注册监听器 } } diff --git a/frontend/utils/socket/SocketManager.js b/frontend/utils/socket/SocketManager.js index c3260a6..ebd9d4b 100644 --- a/frontend/utils/socket/SocketManager.js +++ b/frontend/utils/socket/SocketManager.js @@ -21,6 +21,7 @@ class SocketManager { this.isConnected = false this.isAuthed = false this.isClosing = false // 标记是否主动关闭 + this.pubsubStatus = 'connected' // 'connected' | 'pubsub_disabled'（后端 auth_response 下发） // 事件处理器 this.eventHandlers = { @@ -28,6 +29,7 @@ class SocketManager { 'disconnect': [], 'auth_success': [], 'auth_fail': [], + 'pubsub_disabled': [], // 后端上报 pub/sub 不可用 'error': [], 'message': [] // 通用消息处理 } @@ -107,8 +109,22 @@ class SocketManager { this._isConnecting = true console.log(`[${this.serviceName}] _doConnect called, clearing old socket`) - // 清理旧连接 + // 清除上一次连接的超时定时器，防止跨连接干扰 + if (this._connectTimeout) { + clearTimeout(this._connectTimeout) + this._connectTimeout = null + } + // 清理旧连接：先关闭再丢弃，防止旧 socket 的回调污染状态 if (this.socket) { + try { + if (typeof this.socket.close === 'function') { + this.socket.close() + } else if (typeof this.socket.complete === 'function') { + this.socket.complete() + } + } catch (e) { + console.warn(`[${this.serviceName}] Error closing old socket:`, e) + } this.socket = null } this.isConnected = false @@ -153,10 +169,22 @@ class SocketManager { const socket = this.socket const self = this + // 连接超时保护：15s 内未连接成功则重置 _isConnecting，防止永久阻塞 + // 存储在实例上而非局部变量，防止旧连接的定时器在新连接建立后误触发 + self._connectTimeout = setTimeout(() => { + if (self._isConnecting) { + console.warn(`[${self.serviceName}] Connection timeout (15s), resetting _isConnecting`) + self._isConnecting = false + self._connectTimeout = null + self._emit('error', { code: 'CONNECT_TIMEOUT', message: '连接超时' }) + } + }, 15000) + // 连接打开 if (typeof socket.onOpen === 'function') { socket.onOpen(function() { console.log(`[${self.serviceName}] WebSocket connected`) + clearTimeout(self._connectTimeout); self._connectTimeout = null self.isConnected = true self._isConnecting = false // 连接成功,允许后续重连 // 清除重连计时器 @@ -171,6 +199,7 @@ class SocketManager { // 标准 WebSocket 风格 socket.onopen(function() { console.log(`[${self.serviceName}] WebSocket connected`) + clearTimeout(self._connectTimeout); self._connectTimeout = null self.isConnected = true self._isConnecting = false // 连接成功,允许后续重连 // 清除重连计时器 @@ -202,6 +231,7 @@ class SocketManager { if (typeof socket.onClose === 'function') { socket.onClose(function() { console.log(`[${self.serviceName}] WebSocket closed`) + clearTimeout(self._connectTimeout); self._connectTimeout = null self._isConnecting = false // 连接关闭,允许后续重连 self._cleanup() self._emit('disconnect') @@ -210,6 +240,7 @@ class SocketManager { } else if (typeof socket.onclose === 'function') { socket.onclose(function() { console.log(`[${self.serviceName}] WebSocket closed`) + clearTimeout(self._connectTimeout); self._connectTimeout = null self._isConnecting = false // 连接关闭,允许后续重连 self._cleanup() self._emit('disconnect') @@ -218,10 +249,13 @@ class SocketManager { } // 连接错误 - var handleSocketError = function(err) { + let handleSocketError = function(err) { console.error(`[${self.serviceName}] WebSocket error:`, err) + clearTimeout(self._connectTimeout); self._connectTimeout = null + // 重置连接中状态，避免后续 connect() 调用被永久阻塞 + self._isConnecting = false // 检查是否是鉴权相关的错误（401/403） - var errMsg = (err && (err.errMsg || err.message || '')).toLowerCase() + let errMsg = (err && (err.errMsg || err.message || '')).toLowerCase() if (errMsg.indexOf('auth') !== -1 || errMsg.indexOf('reject') !== -1 || errMsg.indexOf('401') !== -1) { console.warn(`[${self.serviceName}] Connection rejected (auth failure), clearing token`) self._emit('auth_fail', err) @@ -248,6 +282,12 @@ class SocketManager { if (type === 'auth_response') { if (data.success) { this.isAuthed = true + // 记录后端上报的 pub/sub 状态，用于判断是否降级到轮询 + this.pubsubStatus = data.pubsub_status || 'connected' + if (this.pubsubStatus === 'pubsub_disabled') { + console.warn(`[${this.serviceName}] Pub/Sub disabled by server, real-time push unavailable`) + this._emit('pubsub_disabled', data) + } this._emit('auth_success', data) this._startHeartbeat() } else { @@ -347,8 +387,20 @@ class SocketManager { clearTimeout(this.reconnectTimer) this.reconnectTimer = null } + if (this._connectTimeout) { + clearTimeout(this._connectTimeout) + this._connectTimeout = null + } this.isConnected = false this.isAuthed = false + this.pubsubStatus = 'connected' // 重置为默认值，等下次 auth_response 更新 + } + + /** + * 后端 Pub/Sub 是否可用（用于判断是否降级到 HTTP 轮询） + */ + isPubSubEnabled() { + return this.pubsubStatus !== 'pubsub_disabled' } /**