From dc55125f6efd94f3f8a68d5ee82c1bc20fd8786a Mon Sep 17 00:00:00 2001 From: qian Date: Sat, 16 May 2026 21:37:53 +0800 Subject: [PATCH] =?UTF-8?q?refactor(notify):=E9=87=8D=E6=9E=84=E7=AB=99?= =?UTF-8?q?=E5=86=85=E4=BF=A1=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/tenant/tenant.go | 1 + api/tenant/v1/notification_tenant.go | 7 + internal/cmd/cmd.go | 18 - .../tenant_v1_tenant_notification_delete.go | 14 + internal/handler/ws/broker.go | 163 --------- internal/handler/ws/client.go | 156 --------- internal/handler/ws/handler.go | 103 ------ internal/handler/ws/hub.go | 312 ------------------ internal/handler/ws/message.go | 89 ----- internal/logic/admin/notification.go | 3 + internal/logic/common/notification.go | 40 +-- internal/logic/monitor/alert_event.go | 4 - internal/logic/monitor/alert_notify.go | 13 - internal/logic/monitor/ws_push.go | 61 ---- internal/logic/tenant/notification.go | 42 +++ internal/service/tenant.go | 2 + web/admin/src/utils/websocket.ts | 134 -------- .../src/views/monitor/AlertEventsPage.vue | 12 - web/admin/src/views/monitor/MonitorPage.vue | 19 +- .../src/views/monitor/RealtimeMonitorPage.vue | 24 +- .../src/views/operation/MessagesPage.vue | 97 +++++- web/admin/vite.config.ts | 1 - .../components/common/AnnouncementBanner.vue | 28 +- .../src/composables/useNotificationCount.ts | 88 +++-- web/tenant/src/views/Layout.vue | 14 +- .../views/notification/NotificationsPage.vue | 26 ++ 26 files changed, 288 insertions(+), 1183 deletions(-) create mode 100644 internal/controller/tenant/tenant_v1_tenant_notification_delete.go delete mode 100644 internal/handler/ws/broker.go delete mode 100644 internal/handler/ws/client.go delete mode 100644 internal/handler/ws/handler.go delete mode 100644 internal/handler/ws/hub.go delete mode 100644 internal/handler/ws/message.go delete mode 100644 internal/logic/monitor/ws_push.go delete mode 100644 web/admin/src/utils/websocket.ts diff --git a/api/tenant/tenant.go b/api/tenant/tenant.go index 9619707..17a7dae 100644 --- a/api/tenant/tenant.go +++ b/api/tenant/tenant.go @@ -78,6 +78,7 @@ type ITenantV1 interface { TenantUnreadCount(ctx context.Context, req *v1.TenantUnreadCountReq) (res *v1.TenantUnreadCountRes, err error) TenantMarkRead(ctx context.Context, req *v1.TenantMarkReadReq) (res *v1.TenantMarkReadRes, err error) TenantMarkAllRead(ctx context.Context, req *v1.TenantMarkAllReadReq) (res *v1.TenantMarkAllReadRes, err error) + TenantNotificationDelete(ctx context.Context, req *v1.TenantNotificationDeleteReq) (res *v1.TenantNotificationDeleteRes, err error) TenantNotificationPreferencesGet(ctx context.Context, req *v1.TenantNotificationPreferencesGetReq) (res *v1.TenantNotificationPreferencesGetRes, err error) TenantNotificationPreferencesUpdate(ctx context.Context, req *v1.TenantNotificationPreferencesUpdateReq) (res *v1.TenantNotificationPreferencesUpdateRes, err error) TenantAnnouncements(ctx context.Context, req *v1.TenantAnnouncementsReq) (res *v1.TenantAnnouncementsRes, err error) diff --git a/api/tenant/v1/notification_tenant.go b/api/tenant/v1/notification_tenant.go index 2b5d6fe..38dddd0 100644 --- a/api/tenant/v1/notification_tenant.go +++ b/api/tenant/v1/notification_tenant.go @@ -55,6 +55,13 @@ type TenantMarkAllReadReq struct { type TenantMarkAllReadRes struct{} +type TenantNotificationDeleteReq struct { + g.Meta `path:"/notifications/{id}" method:"delete" mime:"json" tags:"租户控制台-通知" summary:"删除消息"` + Id int64 `json:"id" in:"path" v:"required|min:1"` +} + +type TenantNotificationDeleteRes struct{} + type TenantNotificationPreferencesGetReq struct { g.Meta `path:"/notification-preferences" method:"get" mime:"json" tags:"租户控制台-通知" summary:"通知偏好"` } diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index 1577161..0551a57 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -29,7 +29,6 @@ import ( "github.com/qianfree/team-api/internal/handler/public" "github.com/qianfree/team-api/internal/handler/relay" setupHandler "github.com/qianfree/team-api/internal/handler/setup" - wsHandler "github.com/qianfree/team-api/internal/handler/ws" "github.com/qianfree/team-api/internal/plugin" "github.com/qianfree/team-api/web" ) @@ -74,15 +73,9 @@ var ( // Initialize content filter engine common.InitContentFilter(ctx) - // Initialize WebSocket hub and broker - wsHub := wsHandler.InitHub(ctx) - go wsHub.Run(ctx) - wsHandler.InitWsBroker(ctx) - // Initialize monitoring collector monitor.InitCollector(ctx) monitor.InitRequestTracker() - go monitor.StartMonitorPusher(ctx) // Ensure partitioned tables have current+future partitions if partitionErr := common.EnsurePartitions(ctx); partitionErr != nil { @@ -227,16 +220,6 @@ var ( }) }) - // WebSocket endpoints — 独立于 MiddlewareHandlerResponse，避免升级后中间件干扰 - s.Group("/api/admin", func(g *ghttp.RouterGroup) { - g.Middleware(middleware.AdminAuth) - g.GET("/ws", wsHandler.HandleAdminWS) - }) - s.Group("/api/tenant", func(g *ghttp.RouterGroup) { - g.Middleware(middleware.MaintenanceMode, middleware.TenantAuth) - g.GET("/ws", wsHandler.HandleTenantWS) - }) - // Register route groups s.Group("/api", func(group *ghttp.RouterGroup) { group.Middleware(middleware.MiddlewareHandlerResponse) @@ -299,7 +282,6 @@ var ( defer plugin.Shutdown(ctx) defer common.CloseUsageLogWriter() defer response.CloseErrorLogWriter() - defer wsHandler.Shutdown() s.Run() return nil diff --git a/internal/controller/tenant/tenant_v1_tenant_notification_delete.go b/internal/controller/tenant/tenant_v1_tenant_notification_delete.go new file mode 100644 index 0000000..d658558 --- /dev/null +++ b/internal/controller/tenant/tenant_v1_tenant_notification_delete.go @@ -0,0 +1,14 @@ +package tenant + +import ( + "context" + + "github.com/gogf/gf/v2/errors/gcode" + "github.com/gogf/gf/v2/errors/gerror" + + "github.com/qianfree/team-api/api/tenant/v1" +) + +func (c *ControllerV1) TenantNotificationDelete(ctx context.Context, req *v1.TenantNotificationDeleteReq) (res *v1.TenantNotificationDeleteRes, err error) { + return nil, gerror.NewCode(gcode.CodeNotImplemented) +} diff --git a/internal/handler/ws/broker.go b/internal/handler/ws/broker.go deleted file mode 100644 index 6f43b10..0000000 --- a/internal/handler/ws/broker.go +++ /dev/null @@ -1,163 +0,0 @@ -package ws - -import ( - "context" - "encoding/json" - "strconv" - "strings" - "time" - - "github.com/gogf/gf/v2/frame/g" -) - -const redisWsChannel = "ws:messages" - -// WsBroker 封装 Redis Pub/Sub，用于跨实例 WebSocket 消息分发。 -type WsBroker struct { - hub *Hub -} - -// InitWsBroker 初始化 WebSocket 消息代理并启动 Redis 订阅者。 -// 必须在 InitHub 之后调用。 -func InitWsBroker(ctx context.Context) { - hub := GetHub() - if hub == nil { - g.Log().Warning(ctx, "[WsBroker] hub not initialized, skip broker init") - return - } - // 幂等：多次调用只初始化一次 - go startSubscriber(ctx, hub) - g.Log().Info(ctx, "[WsBroker] initialized") -} - -// --- Publish API（业务逻辑调用）--- - -// PublishToAdmin 向特定管理员推送消息。 -func PublishToAdmin(ctx context.Context, userID int64, channel, action string, payload any) error { - return publish(ctx, &RedisWsMessage{ - UserType: "admin", - Target: "user:" + strconv.FormatInt(userID, 10), - Channel: channel, - Action: action, - }, payload) -} - -// PublishToAllAdmins 向所有在线管理员广播消息。 -func PublishToAllAdmins(ctx context.Context, channel, action string, payload any) error { - return publish(ctx, &RedisWsMessage{ - UserType: "admin", - Target: "all", - Channel: channel, - Action: action, - }, payload) -} - -// PublishToTenantUser 向特定租户用户推送消息。 -func PublishToTenantUser(ctx context.Context, tenantID, userID int64, channel, action string, payload any) error { - return publish(ctx, &RedisWsMessage{ - UserType: "tenant", - Target: "user:" + strconv.FormatInt(tenantID, 10) + ":" + strconv.FormatInt(userID, 10), - Channel: channel, - Action: action, - }, payload) -} - -// PublishToTenantAll 向租户所有在线成员广播消息。 -func PublishToTenantAll(ctx context.Context, tenantID int64, channel, action string, payload any) error { - return publish(ctx, &RedisWsMessage{ - UserType: "tenant", - Target: "tenant_all:" + strconv.FormatInt(tenantID, 10), - Channel: channel, - Action: action, - }, payload) -} - -// --- 内部方法 --- - -func publish(ctx context.Context, msg *RedisWsMessage, payload any) error { - data, err := json.Marshal(payload) - if err != nil { - g.Log().Warningf(ctx, "[WsBroker] marshal payload failed: %v", err) - return err - } - msg.Payload = data - - body, err := json.Marshal(msg) - if err != nil { - g.Log().Warningf(ctx, "[WsBroker] marshal message failed: %v", err) - return err - } - - _, err = g.Redis().Do(ctx, "PUBLISH", redisWsChannel, string(body)) - if err != nil { - g.Log().Warningf(ctx, "[WsBroker] redis publish failed: %v", err) - } - return err -} - -func startSubscriber(ctx context.Context, hub *Hub) { - for { - conn, _, err := g.Redis().Subscribe(ctx, redisWsChannel) - if err != nil { - g.Log().Errorf(ctx, "[WsBroker] subscriber connect failed: %v", err) - time.Sleep(5 * time.Second) - continue - } - - g.Log().Info(ctx, "[WsBroker] Redis subscriber started") - - for { - msg, err := conn.ReceiveMessage(ctx) - if err != nil { - g.Log().Warningf(ctx, "[WsBroker] subscriber recv error: %v", err) - conn.Close(ctx) - time.Sleep(5 * time.Second) - break - } - - handleRedisMessage(ctx, hub, msg.Payload) - } - } -} - -func handleRedisMessage(ctx context.Context, hub *Hub, payload string) { - var redisMsg RedisWsMessage - if err := json.Unmarshal([]byte(payload), &redisMsg); err != nil { - g.Log().Warningf(ctx, "[WsBroker] invalid message: %v", err) - return - } - - wsMsg := &WsMessage{ - Channel: redisMsg.Channel, - Action: redisMsg.Action, - Payload: redisMsg.Payload, - } - - switch { - case redisMsg.Target == "all": - if redisMsg.UserType == "admin" { - hub.DispatchToAdmin(ctx, 0, redisMsg.Channel, wsMsg) - } - - case strings.HasPrefix(redisMsg.Target, "tenant_all:"): - tenantID, _ := strconv.ParseInt(strings.TrimPrefix(redisMsg.Target, "tenant_all:"), 10, 64) - if tenantID > 0 { - hub.DispatchToTenantAll(ctx, tenantID, redisMsg.Channel, wsMsg) - } - - case strings.HasPrefix(redisMsg.Target, "user:"): - if redisMsg.UserType == "admin" { - userID, _ := strconv.ParseInt(strings.TrimPrefix(redisMsg.Target, "user:"), 10, 64) - hub.DispatchToAdmin(ctx, userID, redisMsg.Channel, wsMsg) - } else { - parts := strings.SplitN(strings.TrimPrefix(redisMsg.Target, "user:"), ":", 2) - if len(parts) == 2 { - tenantID, _ := strconv.ParseInt(parts[0], 10, 64) - userID, _ := strconv.ParseInt(parts[1], 10, 64) - if tenantID > 0 && userID > 0 { - hub.DispatchToTenant(ctx, tenantID, userID, redisMsg.Channel, wsMsg) - } - } - } - } -} diff --git a/internal/handler/ws/client.go b/internal/handler/ws/client.go deleted file mode 100644 index 2f63232..0000000 --- a/internal/handler/ws/client.go +++ /dev/null @@ -1,156 +0,0 @@ -package ws - -import ( - "context" - "encoding/json" - "sync/atomic" - "time" - - "github.com/gogf/gf/v2/frame/g" - "github.com/gorilla/websocket" -) - -const ( - clientSendBufferSize = 256 - clientMaxMessageSize = 4096 - clientWriteWait = 10 * time.Second - clientPongWait = 60 * time.Second - clientPingPeriod = (clientPongWait * 9) / 10 -) - -// Client 表示一个 WebSocket 连接。 -type Client struct { - hub *Hub - conn *websocket.Conn - send chan []byte - userType string // "admin" / "tenant" - userID int64 - tenantID int64 // admin 为 0 - done chan struct{} - seq int64 -} - -func newClient(hub *Hub, conn *websocket.Conn, userType string, userID, tenantID int64) *Client { - return &Client{ - hub: hub, - conn: conn, - send: make(chan []byte, clientSendBufferSize), - userType: userType, - userID: userID, - tenantID: tenantID, - done: make(chan struct{}), - } -} - -// readPump 从 WebSocket 读取客户端消息。 -// 处理 ping 心跳，退出时触发 close 清理。 -func (c *Client) readPump() { - defer c.close() - - c.conn.SetReadLimit(clientMaxMessageSize) - c.conn.SetReadDeadline(time.Now().Add(clientPongWait)) - c.conn.SetPongHandler(func(string) error { - c.conn.SetReadDeadline(time.Now().Add(clientPongWait)) - return nil - }) - - for { - _, message, err := c.conn.ReadMessage() - if err != nil { - if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { - g.Log().Debugf(context.Background(), - "[WS] read error: type=%s user=%d err=%v", c.userType, c.userID, err) - } - return - } - - var msg WsMessage - if err := json.Unmarshal(message, &msg); err != nil { - continue - } - - switch msg.Action { - case ActionPing: - c.conn.SetReadDeadline(time.Now().Add(clientPongWait)) - pong, _ := json.Marshal(&WsMessage{ - Channel: "", - Action: ActionPong, - }) - select { - case c.send <- pong: - default: - } - } - } -} - -// writePump 将 send channel 中的消息写入 WebSocket。 -// 同时处理定时 ping 和 context 取消时的关闭。 -func (c *Client) writePump(ctx context.Context) { - ticker := time.NewTicker(clientPingPeriod) - defer func() { - ticker.Stop() - c.conn.Close() - }() - - for { - select { - case <-ctx.Done(): - c.conn.SetWriteDeadline(time.Now().Add(clientWriteWait)) - c.conn.WriteMessage(websocket.CloseMessage, - websocket.FormatCloseMessage(websocket.CloseGoingAway, "server shutdown")) - return - - case message, ok := <-c.send: - c.conn.SetWriteDeadline(time.Now().Add(clientWriteWait)) - if !ok { - c.conn.WriteMessage(websocket.CloseMessage, - websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) - return - } - if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { - return - } - - case <-ticker.C: - c.conn.SetWriteDeadline(time.Now().Add(clientWriteWait)) - if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { - return - } - - case <-c.done: - return - } - } -} - -// sendMessage 将 WsMessage 序列化并发送到 send channel。 -// 返回 false 表示客户端缓冲区满（慢客户端）。 -func (c *Client) sendMessage(msg *WsMessage) bool { - msg.Seq = atomic.AddInt64(&c.seq, 1) - data, err := json.Marshal(msg) - if err != nil { - return true - } - select { - case c.send <- data: - return true - default: - return false - } -} - -// close 通知 writePump 退出并从 Hub 注销。 -func (c *Client) close() { - select { - case <-c.done: - // already closed - default: - close(c.done) - } - // 非阻塞注销 - select { - case c.hub.unregister <- c: - default: - } -} diff --git a/internal/handler/ws/handler.go b/internal/handler/ws/handler.go deleted file mode 100644 index 82f832b..0000000 --- a/internal/handler/ws/handler.go +++ /dev/null @@ -1,103 +0,0 @@ -package ws - -import ( - "context" - "net/http" - - "github.com/gogf/gf/v2/frame/g" - "github.com/gogf/gf/v2/net/ghttp" - "github.com/gorilla/websocket" -) - -// Context keys — 与 middleware 包中的常量值保持一致。 -// 不直接导入 middleware 包，避免循环依赖（middleware → logic/common → handler/ws）。 -const ( - ctxKeyUserID = "userId" - ctxKeyTenantID = "tenantId" -) - -func getUserID(ctx context.Context) int64 { - val := ctx.Value(ctxKeyUserID) - if val == nil { - return 0 - } - if id, ok := val.(int64); ok { - return id - } - return 0 -} - -func getTenantID(ctx context.Context) int64 { - val := ctx.Value(ctxKeyTenantID) - if val == nil { - return 0 - } - if id, ok := val.(int64); ok { - return id - } - return 0 -} - -var wsUpgrader = websocket.Upgrader{ - ReadBufferSize: 1024, - WriteBufferSize: 1024, - CheckOrigin: func(r *http.Request) bool { return true }, -} - -// HandleAdminWS 处理 /api/admin/ws WebSocket 连接。 -// AdminAuth 中间件已完成 JWT 验证，context 中已设置用户信息。 -func HandleAdminWS(r *ghttp.Request) { - userID := getUserID(r.Context()) - if userID == 0 { - r.Response.WriteStatus(401, "unauthorized") - return - } - - hub := GetHub() - if hub == nil { - r.Response.WriteStatus(503, "service unavailable") - return - } - - conn, err := wsUpgrader.Upgrade(r.Response.Writer, r.Request, nil) - if err != nil { - g.Log().Warningf(r.Context(), "[WS] admin upgrade failed: user=%d err=%v", userID, err) - return - } - - client := newClient(hub, conn, "admin", userID, 0) - hub.register <- client - - go client.writePump(r.Context()) - client.readPump() -} - -// HandleTenantWS 处理 /api/tenant/ws WebSocket 连接。 -// TenantAuth 中间件已完成 JWT 验证，context 中已设置租户和用户信息。 -func HandleTenantWS(r *ghttp.Request) { - tenantID := getTenantID(r.Context()) - userID := getUserID(r.Context()) - if userID == 0 || tenantID == 0 { - r.Response.WriteStatus(401, "unauthorized") - return - } - - hub := GetHub() - if hub == nil { - r.Response.WriteStatus(503, "service unavailable") - return - } - - conn, err := wsUpgrader.Upgrade(r.Response.Writer, r.Request, nil) - if err != nil { - g.Log().Warningf(r.Context(), "[WS] tenant upgrade failed: tenant=%d user=%d err=%v", - tenantID, userID, err) - return - } - - client := newClient(hub, conn, "tenant", userID, tenantID) - hub.register <- client - - go client.writePump(r.Context()) - client.readPump() -} diff --git a/internal/handler/ws/hub.go b/internal/handler/ws/hub.go deleted file mode 100644 index 9fb6923..0000000 --- a/internal/handler/ws/hub.go +++ /dev/null @@ -1,312 +0,0 @@ -package ws - -import ( - "context" - "sync" - - "github.com/gogf/gf/v2/frame/g" -) - -// dispatchRequest 是 Hub 内部消息路由请求。 -type dispatchRequest struct { - userType string // "admin" / "tenant" - userID int64 // 0 = 广播 - tenantID int64 // 租户广播时使用 - channel string - msg *WsMessage -} - -// Hub 管理所有活跃的 WebSocket 连接，并按用户路由消息。 -// 每个服务实例运行一个 Hub。 -type Hub struct { - adminClients map[int64]map[*Client]bool // userID → clients（支持多标签页） - tenantClients map[string]map[*Client]bool // "tenantID:userID" → clients - allClients map[*Client]bool - - register chan *Client - unregister chan *Client - dispatch chan *dispatchRequest - - mu sync.RWMutex -} - -var ( - defaultHub *Hub - defaultHubOnce sync.Once -) - -// InitHub 创建并返回 Hub 单例。 -func InitHub(_ context.Context) *Hub { - defaultHubOnce.Do(func() { - defaultHub = &Hub{ - adminClients: make(map[int64]map[*Client]bool), - tenantClients: make(map[string]map[*Client]bool), - allClients: make(map[*Client]bool), - register: make(chan *Client, 64), - unregister: make(chan *Client, 64), - dispatch: make(chan *dispatchRequest, 512), - } - }) - return defaultHub -} - -// GetHub 返回 Hub 单例，未初始化时返回 nil。 -func GetHub() *Hub { - return defaultHub -} - -// Run 启动 Hub 事件循环，阻塞直到 context 取消。 -func (h *Hub) Run(ctx context.Context) { - g.Log().Info(ctx, "[WS Hub] started") - defer g.Log().Info(ctx, "[WS Hub] stopped") - - for { - select { - case <-ctx.Done(): - h.drainAll() - return - - case client := <-h.register: - h.registerClient(client) - - case client := <-h.unregister: - h.unregisterClient(client) - - case req := <-h.dispatch: - h.handleDispatch(req) - } - } -} - -// Shutdown 立即关闭所有连接。 -func Shutdown() { - h := GetHub() - if h == nil { - return - } - h.mu.Lock() - for client := range h.allClients { - client.close() - } - h.mu.Unlock() -} - -// DispatchToAdmin 向管理后台用户推送消息。userID=0 时广播给所有管理员。 -func (h *Hub) DispatchToAdmin(_ context.Context, userID int64, channel string, msg *WsMessage) { - select { - case h.dispatch <- &dispatchRequest{ - userType: "admin", - userID: userID, - channel: channel, - msg: msg, - }: - default: - g.Log().Warningf(context.Background(), - "[WS Hub] dispatch channel full, dropping admin msg: user=%d channel=%s", userID, channel) - } -} - -// DispatchToTenant 向特定租户用户推送消息。 -func (h *Hub) DispatchToTenant(_ context.Context, tenantID, userID int64, channel string, msg *WsMessage) { - select { - case h.dispatch <- &dispatchRequest{ - userType: "tenant", - userID: userID, - tenantID: tenantID, - channel: channel, - msg: msg, - }: - default: - g.Log().Warningf(context.Background(), - "[WS Hub] dispatch channel full, dropping tenant msg: tenant=%d user=%d channel=%s", - tenantID, userID, channel) - } -} - -// DispatchToTenantAll 向租户所有在线成员推送消息。 -func (h *Hub) DispatchToTenantAll(_ context.Context, tenantID int64, channel string, msg *WsMessage) { - select { - case h.dispatch <- &dispatchRequest{ - userType: "tenant", - userID: 0, - tenantID: tenantID, - channel: channel, - msg: msg, - }: - default: - g.Log().Warningf(context.Background(), - "[WS Hub] dispatch channel full, dropping tenant broadcast: tenant=%d channel=%s", - tenantID, channel) - } -} - -// ClientCount 返回当前活跃连接数。 -func (h *Hub) ClientCount() int { - h.mu.RLock() - defer h.mu.RUnlock() - return len(h.allClients) -} - -// --- 内部方法 --- - -func (h *Hub) registerClient(client *Client) { - h.mu.Lock() - defer h.mu.Unlock() - - h.allClients[client] = true - - if client.userType == "admin" { - if h.adminClients[client.userID] == nil { - h.adminClients[client.userID] = make(map[*Client]bool) - } - h.adminClients[client.userID][client] = true - } else { - key := tenantClientKey(client.tenantID, client.userID) - if h.tenantClients[key] == nil { - h.tenantClients[key] = make(map[*Client]bool) - } - h.tenantClients[key][client] = true - } - - g.Log().Debugf(context.Background(), - "[WS Hub] client registered: type=%s user=%d tenant=%d total=%d", - client.userType, client.userID, client.tenantID, len(h.allClients)) -} - -func (h *Hub) unregisterClient(client *Client) { - h.mu.Lock() - defer h.mu.Unlock() - - if !h.allClients[client] { - return - } - - delete(h.allClients, client) - - if client.userType == "admin" { - if clients, ok := h.adminClients[client.userID]; ok { - delete(clients, client) - if len(clients) == 0 { - delete(h.adminClients, client.userID) - } - } - } else { - key := tenantClientKey(client.tenantID, client.userID) - if clients, ok := h.tenantClients[key]; ok { - delete(clients, client) - if len(clients) == 0 { - delete(h.tenantClients, key) - } - } - } - - // 排空 send channel，释放消息内存 - drained := 0 -drain: - for { - select { - case <-client.send: - drained++ - default: - break drain - } - } - close(client.send) - - g.Log().Debugf(context.Background(), - "[WS Hub] client unregistered: type=%s user=%d drained=%d total=%d", - client.userType, client.userID, drained, len(h.allClients)) -} - -func (h *Hub) handleDispatch(req *dispatchRequest) { - h.mu.RLock() - defer h.mu.RUnlock() - - switch req.userType { - case "admin": - if req.userID > 0 { - // 特定管理员 - for client := range h.adminClients[req.userID] { - if !client.sendMessage(req.msg) { - go client.close() - } - } - } else { - // 广播所有管理员 - for _, clients := range h.adminClients { - for client := range clients { - if !client.sendMessage(req.msg) { - go client.close() - } - } - } - } - - case "tenant": - if req.userID > 0 { - // 特定租户用户 - key := tenantClientKey(req.tenantID, req.userID) - for client := range h.tenantClients[key] { - if !client.sendMessage(req.msg) { - go client.close() - } - } - } else if req.tenantID > 0 { - // 租户广播：匹配 "tenantID:*" 的所有用户 - prefix := tenantClientKeyPrefix(req.tenantID) - for key, clients := range h.tenantClients { - if len(key) >= len(prefix) && key[:len(prefix)] == prefix { - for client := range clients { - if !client.sendMessage(req.msg) { - go client.close() - } - } - } - } - } - } -} - -func (h *Hub) drainAll() { - h.mu.Lock() - defer h.mu.Unlock() - - for client := range h.allClients { - client.close() - } - h.adminClients = make(map[int64]map[*Client]bool) - h.tenantClients = make(map[string]map[*Client]bool) - h.allClients = make(map[*Client]bool) -} - -func tenantClientKey(tenantID, userID int64) string { - return formatInt64(tenantID) + ":" + formatInt64(userID) -} - -func tenantClientKeyPrefix(tenantID int64) string { - return formatInt64(tenantID) + ":" -} - -func formatInt64(v int64) string { - if v == 0 { - return "0" - } - // 快速 int64 → string，避免 strconv 依赖 - buf := make([]byte, 0, 20) - neg := v < 0 - if neg { - v = -v - } - for v > 0 { - buf = append(buf, byte('0'+v%10)) - v /= 10 - } - if neg { - buf = append(buf, '-') - } - // 反转 - for i, j := 0, len(buf)-1; i < j; i, j = i+1, j-1 { - buf[i], buf[j] = buf[j], buf[i] - } - return string(buf) -} diff --git a/internal/handler/ws/message.go b/internal/handler/ws/message.go deleted file mode 100644 index 6c36fdf..0000000 --- a/internal/handler/ws/message.go +++ /dev/null @@ -1,89 +0,0 @@ -package ws - -import "encoding/json" - -// 频道常量 -const ( - ChannelNotification = "notification" - ChannelAlert = "alert" - ChannelMonitor = "monitor" - ChannelTask = "task" - ChannelBilling = "billing" -) - -// 客户端 action -const ( - ActionPing = "ping" -) - -// 服务端 action -const ( - ActionPong = "pong" - ActionCreated = "created" - ActionUpdated = "updated" - ActionResolved = "resolved" - ActionProgress = "progress" - ActionSnapshot = "snapshot" -) - -// WsMessage 是所有 WebSocket 消息的统一信封。 -// 客户端→服务端：Action 为 ping。 -// 服务端→客户端：Action 为 created / updated / progress 等。 -type WsMessage struct { - Channel string `json:"channel"` - Action string `json:"action"` - Payload json.RawMessage `json:"payload,omitempty"` - Seq int64 `json:"seq,omitempty"` -} - -// RedisWsMessage 是 Redis Pub/Sub 传输信封，携带路由信息。 -type RedisWsMessage struct { - UserType string `json:"user_type"` // "admin" / "tenant" - Target string `json:"target"` // "user:{id}" / "all" / "tenant_all:{tid}" - Channel string `json:"channel"` - Action string `json:"action"` - Payload json.RawMessage `json:"payload"` -} - -// NotificationPayload 通知频道 payload -type NotificationPayload struct { - ID int64 `json:"id"` - Type string `json:"type"` - Title string `json:"title"` - Content string `json:"content"` - IsBroadcast bool `json:"is_broadcast"` - CreatedAt string `json:"created_at"` -} - -// AlertPayload 告警频道 payload -type AlertPayload struct { - ID int64 `json:"id"` - RuleName string `json:"rule_name"` - MetricType string `json:"metric_type"` - Level string `json:"level"` - Status string `json:"status"` - TriggerValue float64 `json:"trigger_value"` - Threshold float64 `json:"threshold"` - TriggerMessage string `json:"trigger_message"` - Timestamp string `json:"timestamp"` -} - -// TaskPayload 任务频道 payload -type TaskPayload struct { - TaskID string `json:"task_id"` - Status string `json:"status"` - Progress int `json:"progress"` - FailReason string `json:"fail_reason,omitempty"` - ResultURL string `json:"result_url,omitempty"` - UpdatedAt string `json:"updated_at"` -} - -// BillingPayload 计费频道 payload -type BillingPayload struct { - EventType string `json:"event_type"` - BalanceBefore float64 `json:"balance_before,omitempty"` - BalanceAfter float64 `json:"balance_after,omitempty"` - Amount float64 `json:"amount,omitempty"` - Message string `json:"message,omitempty"` - Timestamp string `json:"timestamp"` -} diff --git a/internal/logic/admin/notification.go b/internal/logic/admin/notification.go index 083ddcb..4e4fa28 100644 --- a/internal/logic/admin/notification.go +++ b/internal/logic/admin/notification.go @@ -325,6 +325,7 @@ func (s *sAdmin) UpdateAnnouncement(ctx context.Context, req *v1.AnnouncementUpd if err != nil { return nil, err } + return nil, nil } @@ -365,6 +366,7 @@ func (s *sAdmin) PublishAnnouncement(ctx context.Context, req *v1.AnnouncementPu if err != nil { return nil, err } + return nil, nil } @@ -378,6 +380,7 @@ func (s *sAdmin) ArchiveAnnouncement(ctx context.Context, req *v1.AnnouncementAr if err != nil { return nil, err } + return nil, nil } diff --git a/internal/logic/common/notification.go b/internal/logic/common/notification.go index dc51d1b..6c81d11 100644 --- a/internal/logic/common/notification.go +++ b/internal/logic/common/notification.go @@ -2,12 +2,9 @@ package common import ( "context" - "database/sql" "encoding/json" - "time" "github.com/qianfree/team-api/internal/dao" - ws "github.com/qianfree/team-api/internal/handler/ws" do "github.com/qianfree/team-api/internal/model/do" "github.com/gogf/gf/v2/errors/gerror" @@ -79,7 +76,7 @@ func (e *NotificationEngine) SendNotification(ctx context.Context, tenantID, use // 6. Create in-app message if sendInApp { - result, insertErr := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ + _, insertErr := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ TenantId: tenantID, UserId: userID, Type: category, @@ -92,8 +89,7 @@ func (e *NotificationEngine) SendNotification(ctx context.Context, tenantID, use }) if insertErr != nil { g.Log().Warningf(ctx, "notification engine: create in-app message failed: %v", insertErr) - } else { - go e.pushNotificationToWs(ctx, tenantID, userID, result, category, subject, body, false) + } } @@ -126,7 +122,7 @@ func (e *NotificationEngine) SendBroadcast(ctx context.Context, tenantID int64, category := e.templateCategory(tpl.Channel, templateCode) // 3. Create broadcast in-app message (user_id=NULL) - result, err := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ + _, err = dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ TenantId: tenantID, UserId: nil, Type: category, @@ -141,7 +137,6 @@ func (e *NotificationEngine) SendBroadcast(ctx context.Context, tenantID int64, if err != nil { return gerror.Wrapf(err, "create broadcast message") } - go e.pushNotificationToWs(ctx, tenantID, 0, result, category, subject, body, true) // 4. Send email to all tenant members (async, best-effort) go e.broadcastEmails(ctx, tenantID, subject, body, templateCode, variables) @@ -180,7 +175,7 @@ func (e *NotificationEngine) SendToAllTenants(ctx context.Context, templateCode // SendMessage creates a direct in-app message without a template. func (e *NotificationEngine) SendMessage(ctx context.Context, tenantID, userID int64, msgType, title, content string) error { - result, err := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ + _, err := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ TenantId: tenantID, UserId: userID, Type: msgType, @@ -191,9 +186,6 @@ func (e *NotificationEngine) SendMessage(ctx context.Context, tenantID, userID i IsBroadcast: 0, Metadata: nil, }) - if err == nil { - go e.pushNotificationToWs(ctx, tenantID, userID, result, msgType, title, content, false) - } return err } @@ -207,7 +199,7 @@ func (e *NotificationEngine) SendBroadcastMessage(ctx context.Context, tenantID targetRolesVal = targetRoles } - result, err := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ + _, err := dao.NtfMessages.Ctx(ctx).Insert(do.NtfMessages{ TenantId: tenantID, UserId: nil, Type: msgType, @@ -219,9 +211,6 @@ func (e *NotificationEngine) SendBroadcastMessage(ctx context.Context, tenantID Metadata: nil, TargetRoles: targetRolesVal, }) - if err == nil { - go e.pushNotificationToWs(ctx, tenantID, 0, result, msgType, title, content, true) - } return err } @@ -445,25 +434,6 @@ func (e *NotificationEngine) logNotification(ctx context.Context, tenantID, user } } -// pushNotificationToWs ͨ�� WebSocket ����֪ͨ��Ϣ�������û��� -// isBroadcast=true ʱ���͸��⻧���г�Ա���������͸��ض��û��� -func (e *NotificationEngine) pushNotificationToWs(ctx context.Context, tenantID, userID int64, result sql.Result, nType, title, body string, isBroadcast bool) { - id, _ := result.LastInsertId() - payload := ws.NotificationPayload{ - ID: id, - Type: nType, - Title: title, - Content: body, - IsBroadcast: isBroadcast, - CreatedAt: time.Now().Format(time.RFC3339), - } - if isBroadcast { - ws.PublishToTenantAll(ctx, tenantID, ws.ChannelNotification, ws.ActionCreated, payload) - } else { - ws.PublishToTenantUser(ctx, tenantID, userID, ws.ChannelNotification, ws.ActionCreated, payload) - } -} - // containsAny checks if the haystack contains any of the given substrings. func containsAny(haystack string, needles ...string) bool { for _, n := range needles { diff --git a/internal/logic/monitor/alert_event.go b/internal/logic/monitor/alert_event.go index 920d07e..b93f5f8 100644 --- a/internal/logic/monitor/alert_event.go +++ b/internal/logic/monitor/alert_event.go @@ -6,7 +6,6 @@ import ( "time" "github.com/qianfree/team-api/internal/dao" - ws "github.com/qianfree/team-api/internal/handler/ws" do "github.com/qianfree/team-api/internal/model/do" "github.com/gogf/gf/v2/frame/g" @@ -104,7 +103,6 @@ func AcknowledgeAlert(ctx context.Context, eventID, adminID int64) error { return lcommon.NewBusinessError(404, "事件不存在或状态不是 firing") } - go ws.PublishToAllAdmins(ctx, ws.ChannelAlert, ws.ActionUpdated, map[string]any{"id": eventID, "status": "acknowledged"}) return nil } @@ -128,8 +126,6 @@ func ResolveAlert(ctx context.Context, eventID, adminID int64, notes string) err return lcommon.NewBusinessError(404, "事件不存在或已解决") } - go ws.PublishToAllAdmins(ctx, ws.ChannelAlert, ws.ActionUpdated, map[string]any{"id": eventID, "status": "resolved"}) - // Clear the firing state in Redis for this rule type eventRow struct { RuleID int64 `json:"rule_id"` diff --git a/internal/logic/monitor/alert_notify.go b/internal/logic/monitor/alert_notify.go index a35222e..41fd410 100644 --- a/internal/logic/monitor/alert_notify.go +++ b/internal/logic/monitor/alert_notify.go @@ -10,7 +10,6 @@ import ( "github.com/gogf/gf/v2/util/gconv" "github.com/qianfree/team-api/internal/dao" - ws "github.com/qianfree/team-api/internal/handler/ws" "github.com/qianfree/team-api/internal/logic/common" do "github.com/qianfree/team-api/internal/model/do" ) @@ -59,18 +58,6 @@ func dispatchAlertNotifications(ctx context.Context, rule map[string]any, eventI Update() } - // Push alert via WebSocket to all admin connections - go ws.PublishToAllAdmins(ctx, ws.ChannelAlert, ws.ActionCreated, ws.AlertPayload{ - ID: eventID, - RuleName: ruleName, - MetricType: metricType, - Level: level, - Status: "firing", - TriggerValue: triggerValue, - Threshold: threshold, - TriggerMessage: message, - Timestamp: time.Now().Format(time.RFC3339), - }) } // sendAlertEmailToAdmins sends alert email to specified admin users. diff --git a/internal/logic/monitor/ws_push.go b/internal/logic/monitor/ws_push.go deleted file mode 100644 index 119f0e1..0000000 --- a/internal/logic/monitor/ws_push.go +++ /dev/null @@ -1,61 +0,0 @@ -package monitor - -import ( - "context" - "time" - - "github.com/gogf/gf/v2/frame/g" - - ws "github.com/qianfree/team-api/internal/handler/ws" -) - -// StartMonitorPusher 启动监控数据 WebSocket 推送服务。 -// realtime 数据每 3 秒推送一次，dashboard 数据每 30 秒推送一次。 -// 仅当有 admin WS 连接时才执行数据采集和推送。 -func StartMonitorPusher(ctx context.Context) { - go realtimePusher(ctx) - go dashboardPusher(ctx) - g.Log().Info(ctx, "[MonitorPusher] started") -} - -func realtimePusher(ctx context.Context) { - ticker := time.NewTicker(3 * time.Second) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - hub := ws.GetHub() - if hub == nil || hub.ClientCount() == 0 { - continue - } - data := GetRealtimeData() - ws.PublishToAllAdmins(ctx, ws.ChannelMonitor, "realtime", data) - } - } -} - -func dashboardPusher(ctx context.Context) { - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - hub := ws.GetHub() - if hub == nil || hub.ClientCount() == 0 { - continue - } - data, err := GetDashboardData(ctx, 5) - if err != nil { - g.Log().Warningf(ctx, "[MonitorPusher] dashboard data error: %v", err) - continue - } - ws.PublishToAllAdmins(ctx, ws.ChannelMonitor, "dashboard", data) - } - } -} diff --git a/internal/logic/tenant/notification.go b/internal/logic/tenant/notification.go index 6c3ac79..541d337 100644 --- a/internal/logic/tenant/notification.go +++ b/internal/logic/tenant/notification.go @@ -231,6 +231,48 @@ func (s *sTenant) MarkAllRead(ctx context.Context, req *v1.TenantMarkAllReadReq) return &v1.TenantMarkAllReadRes{}, nil } +// DeleteNotification 删除已读的个人消息（广播消息不允许用户删除） +func (s *sTenant) DeleteNotification(ctx context.Context, req *v1.TenantNotificationDeleteReq) (*v1.TenantNotificationDeleteRes, error) { + tenantID := ctxTenantID(ctx) + userID := ctxUserID(ctx) + + var msg struct { + ID int64 `json:"id"` + IsBroadcast int `json:"is_broadcast"` + UserID *int64 `json:"user_id"` + IsRead int `json:"is_read"` + } + err := dao.NtfMessages.Ctx(ctx). + Where("id", req.Id). + Where("tenant_id", tenantID). + Scan(&msg) + if err != nil { + return nil, err + } + if msg.ID == 0 { + return nil, common.NewNotFoundError("消息") + } + if msg.IsBroadcast == 1 { + return nil, common.NewBadRequestError("广播消息不支持删除") + } + if msg.UserID == nil || *msg.UserID != userID { + return nil, common.NewForbiddenError("无权删除该消息") + } + if msg.IsRead == 0 { + return nil, common.NewBadRequestError("只能删除已读消息") + } + + _, err = dao.NtfMessages.Ctx(ctx). + Where("id", req.Id). + Where("tenant_id", tenantID). + Delete() + if err != nil { + return nil, err + } + + return &v1.TenantNotificationDeleteRes{}, nil +} + // NotificationPreferencesGet 获取合并后的通知偏好（组织 + 用户） func (s *sTenant) NotificationPreferencesGet(ctx context.Context, req *v1.TenantNotificationPreferencesGetReq) (*v1.TenantNotificationPreferencesGetRes, error) { tenantID := ctxTenantID(ctx) diff --git a/internal/service/tenant.go b/internal/service/tenant.go index 149980c..0542299 100644 --- a/internal/service/tenant.go +++ b/internal/service/tenant.go @@ -157,6 +157,8 @@ type ( MarkRead(ctx context.Context, req *v1.TenantMarkReadReq) (*v1.TenantMarkReadRes, error) // MarkAllRead 标记所有未读消息为已读 MarkAllRead(ctx context.Context, req *v1.TenantMarkAllReadReq) (*v1.TenantMarkAllReadRes, error) + // DeleteNotification 删除已读的个人消息（广播消息不允许用户删除） + DeleteNotification(ctx context.Context, req *v1.TenantNotificationDeleteReq) (*v1.TenantNotificationDeleteRes, error) // NotificationPreferencesGet 获取合并后的通知偏好（组织 + 用户） NotificationPreferencesGet(ctx context.Context, req *v1.TenantNotificationPreferencesGetReq) (*v1.TenantNotificationPreferencesGetRes, error) // NotificationPreferencesUpdate 更新通知偏好 diff --git a/web/admin/src/utils/websocket.ts b/web/admin/src/utils/websocket.ts deleted file mode 100644 index 31dc6e3..0000000 --- a/web/admin/src/utils/websocket.ts +++ /dev/null @@ -1,134 +0,0 @@ -import { ref } from 'vue' -import { getAccessToken } from './request' - -export interface WsMessage { - channel: string - action: string - payload?: any - seq?: number -} - -type MessageHandler = (msg: WsMessage) => void - -const wsConnected = ref(false) - -class WsManager { - private ws: WebSocket | null = null - private handlers = new Map>() - private reconnectTimer: ReturnType | null = null - private reconnectDelay = 1000 - private maxReconnectDelay = 30000 - private disposed = false - - connect(): void { - if (this.ws?.readyState === WebSocket.OPEN || this.ws?.readyState === WebSocket.CONNECTING) { - return - } - - const token = getAccessToken() - if (!token) return - - const protocol = location.protocol === 'https:' ? 'wss:' : 'ws:' - const url = `${protocol}//${location.host}/api/admin/ws?token=${encodeURIComponent(token)}` - this.ws = new WebSocket(url) - this.disposed = false - - this.ws.onopen = () => { - this.reconnectDelay = 1000 - wsConnected.value = true - } - - this.ws.onmessage = (event) => { - try { - const msg: WsMessage = JSON.parse(event.data) - const channelHandlers = this.handlers.get(msg.channel) - if (channelHandlers) { - channelHandlers.forEach((handler) => handler(msg)) - } - // Also notify wildcard handlers - const wildcardHandlers = this.handlers.get('*') - if (wildcardHandlers) { - wildcardHandlers.forEach((handler) => handler(msg)) - } - } catch { - // ignore malformed messages - } - } - - this.ws.onclose = () => { - wsConnected.value = false - this.scheduleReconnect() - } - - this.ws.onerror = () => { - // onclose will fire after this - } - } - - private scheduleReconnect(): void { - if (this.disposed) return - if (this.reconnectTimer) return - - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null - this.connect() - }, this.reconnectDelay) - - this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay) - } - - disconnect(): void { - this.disposed = true - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer) - this.reconnectTimer = null - } - if (this.ws) { - this.ws.onclose = null - this.ws.close() - this.ws = null - } - } - - on(channel: string, handler: MessageHandler): void { - if (!this.handlers.has(channel)) { - this.handlers.set(channel, new Set()) - } - this.handlers.get(channel)!.add(handler) - } - - off(channel: string, handler: MessageHandler): void { - const channelHandlers = this.handlers.get(channel) - if (channelHandlers) { - channelHandlers.delete(handler) - if (channelHandlers.size === 0) { - this.handlers.delete(channel) - } - } - } - - get isConnected(): boolean { - return this.ws?.readyState === WebSocket.OPEN - } -} - -// Singleton instance -export const wsManager = new WsManager() - -// Composable for Vue components -export function useWebSocket(): { - on: (channel: string, handler: MessageHandler) => void - off: (channel: string, handler: MessageHandler) => void - connected: typeof wsConnected -} { - // Auto-connect on first use if not already connected - if (!wsManager.isConnected) { - wsManager.connect() - } - - return { - on: (channel: string, handler: MessageHandler) => wsManager.on(channel, handler), - off: (channel: string, handler: MessageHandler) => wsManager.off(channel, handler), - connected: wsConnected, - } -} diff --git a/web/admin/src/views/monitor/AlertEventsPage.vue b/web/admin/src/views/monitor/AlertEventsPage.vue index 7945858..41dfc32 100644 --- a/web/admin/src/views/monitor/AlertEventsPage.vue +++ b/web/admin/src/views/monitor/AlertEventsPage.vue @@ -1,7 +1,5 @@ diff --git a/web/admin/src/views/monitor/MonitorPage.vue b/web/admin/src/views/monitor/MonitorPage.vue index 0c1f98c..9fedeb5 100644 --- a/web/admin/src/views/monitor/MonitorPage.vue +++ b/web/admin/src/views/monitor/MonitorPage.vue @@ -2,8 +2,6 @@ import { ref, onMounted, onUnmounted } from 'vue' import PageHeader from '@/components/PageHeader.vue' import request from '@/utils/request' -import { useWebSocket } from '@/utils/websocket' -import type { WsMessage } from '@/utils/websocket' const loading = ref(false) const timeRange = ref(5) // minutes @@ -36,21 +34,18 @@ function colorThreshold(v: any, warnAt: number, critAt: number): string { return '#00b42a' } -const { on: wsOn, off: wsOff } = useWebSocket() - -function handleMonitorMessage(msg: WsMessage) { - if (msg.action === 'dashboard' && msg.payload) { - dashboardData.value = msg.payload - } -} +let pollTimer: ReturnType | null = null onMounted(() => { - fetchDashboard() // initial load via HTTP - wsOn('monitor', handleMonitorMessage) + fetchDashboard() + pollTimer = setInterval(fetchDashboard, 30000) }) onUnmounted(() => { - wsOff('monitor', handleMonitorMessage) + if (pollTimer) { + clearInterval(pollTimer) + pollTimer = null + } }) diff --git a/web/admin/src/views/monitor/RealtimeMonitorPage.vue b/web/admin/src/views/monitor/RealtimeMonitorPage.vue index 7f4ec92..b9032ea 100644 --- a/web/admin/src/views/monitor/RealtimeMonitorPage.vue +++ b/web/admin/src/views/monitor/RealtimeMonitorPage.vue @@ -7,8 +7,6 @@ import { GridComponent, TooltipComponent, LegendComponent } from 'echarts/compon import VChart from 'vue-echarts' import PageHeader from '@/components/PageHeader.vue' import request from '@/utils/request' -import { useWebSocket } from '@/utils/websocket' -import type { WsMessage } from '@/utils/websocket' use([CanvasRenderer, LineChart, BarChart, GridComponent, TooltipComponent, LegendComponent]) @@ -165,25 +163,19 @@ const heapUsagePercent = computed(() => { return Math.min(100, (inuse / sys) * 100) }) -const { on: wsOn, off: wsOff, connected } = useWebSocket() - -function handleMonitorMessage(msg: WsMessage) { - if (msg.action === 'realtime' && msg.payload) { - data.value = msg.payload - lastUpdated.value = new Date() - updateCharts() - } -} +let pollTimer: ReturnType | null = null onMounted(() => { fetchRealtime() - wsOn('monitor', handleMonitorMessage) - // Update elapsed time every second + pollTimer = setInterval(fetchRealtime, 10000) elapsedTimer = setInterval(() => { now.value = Date.now() }, 1000) }) onUnmounted(() => { - wsOff('monitor', handleMonitorMessage) + if (pollTimer) { + clearInterval(pollTimer) + pollTimer = null + } if (elapsedTimer) { clearInterval(elapsedTimer) elapsedTimer = null @@ -196,8 +188,8 @@ onUnmounted(() => {