refactor(notify):重构站内信功能
This commit is contained in:
qian committed
1 parent
16711597c9
commit
dc55125f6e
26 files changed
+288
-1183
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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:"通知偏好"`
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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 更新通知偏好
|
||||
|
||||
@@ -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<string, Set<MessageHandler>>()
|
||||
private reconnectTimer: ReturnType<typeof setTimeout> | 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,
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,5 @@
|
||||
<script setup lang="ts">
|
||||
import { ref, reactive, onMounted, onUnmounted, h } from 'vue'
|
||||
import { useWebSocket } from '@/utils/websocket'
|
||||
import type { WsMessage } from '@/utils/websocket'
|
||||
import { Message, Tag, Button, Space } from '@arco-design/web-vue'
|
||||
import type { TableColumnData } from '@arco-design/web-vue'
|
||||
import PageHeader from '@/components/PageHeader.vue'
|
||||
@@ -146,22 +144,12 @@ function handlePageChange(page: number) {
|
||||
fetchData()
|
||||
}
|
||||
|
||||
const { on: wsOn, off: wsOff } = useWebSocket()
|
||||
|
||||
function handleAlertMessage(msg: WsMessage) {
|
||||
// Refresh list on any alert event
|
||||
if (msg.action === 'created' || msg.action === 'updated') {
|
||||
fetchData()
|
||||
}
|
||||
}
|
||||
|
||||
onMounted(() => {
|
||||
fetchData()
|
||||
wsOn('alert', handleAlertMessage)
|
||||
})
|
||||
|
||||
onUnmounted(() => {
|
||||
wsOff('alert', handleAlertMessage)
|
||||
})
|
||||
</script>
|
||||
|
||||
|
||||
@@ -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<typeof setInterval> | 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
|
||||
}
|
||||
})
|
||||
</script>
|
||||
|
||||
|
||||
@@ -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<typeof setInterval> | 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(() => {
|
||||
<PageHeader title="实时监控面板" description="并发请求、网络流量与运行时状态实时监控">
|
||||
<template #actions>
|
||||
<div class="header-status">
|
||||
<a-tag :color="connected ? 'green' : 'red'" size="small">
|
||||
{{ connected ? 'WS 已连接' : 'WS 断开' }}
|
||||
<a-tag color="green" size="small">
|
||||
自动刷新 10s
|
||||
</a-tag>
|
||||
<span v-if="lastUpdated" class="last-updated">最后更新 {{ fmtLastUpdated() }}</span>
|
||||
</div>
|
||||
|
||||
@@ -118,12 +118,82 @@ function openTenantDropdown() {
|
||||
}
|
||||
}
|
||||
|
||||
// === User selector (for send modal, depends on tenant) ===
|
||||
const userOptions = ref<{ value: number; label: string }[]>([])
|
||||
const userLoading = ref(false)
|
||||
const userKeyword = ref('')
|
||||
const userPage = ref(1)
|
||||
const userHasMore = ref(false)
|
||||
|
||||
let userSearchTimer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
async function fetchUserOptions(tenantId: number, reset = true) {
|
||||
userLoading.value = true
|
||||
try {
|
||||
const params: any = { page: userPage.value, page_size: 20, tenant_id: tenantId }
|
||||
if (userKeyword.value) params.keyword = userKeyword.value
|
||||
const res = await request.get('/admin/members', { params })
|
||||
const d = res.data?.data
|
||||
const list: { id: number; username: string; display_name: string; email: string }[] = d?.list || []
|
||||
const items = list.map(u => ({ value: u.id, label: u.display_name ? `${u.display_name} (${u.username})` : u.username }))
|
||||
if (reset) {
|
||||
userOptions.value = items
|
||||
} else {
|
||||
userOptions.value = [...userOptions.value, ...items]
|
||||
}
|
||||
const total: number = d?.total || 0
|
||||
userHasMore.value = userOptions.value.length < total
|
||||
} catch {
|
||||
// ignore
|
||||
} finally {
|
||||
userLoading.value = false
|
||||
}
|
||||
}
|
||||
|
||||
function handleUserSearch(val: string) {
|
||||
if (!sendForm.tenant_id) return
|
||||
if (userSearchTimer) clearTimeout(userSearchTimer)
|
||||
userSearchTimer = setTimeout(() => {
|
||||
userKeyword.value = val
|
||||
userPage.value = 1
|
||||
fetchUserOptions(sendForm.tenant_id!, true)
|
||||
}, 300)
|
||||
}
|
||||
|
||||
function handleUserScrollReachEdge(e: Event) {
|
||||
if (!sendForm.tenant_id || !userHasMore.value || userLoading.value) return
|
||||
const el = e.target as HTMLElement
|
||||
const bottom = el.scrollHeight - el.scrollTop - el.clientHeight
|
||||
if (bottom < 30) {
|
||||
userPage.value++
|
||||
fetchUserOptions(sendForm.tenant_id, false)
|
||||
}
|
||||
}
|
||||
|
||||
function handleSendTenantChange(val: number | undefined) {
|
||||
sendForm.user_id = undefined
|
||||
userOptions.value = []
|
||||
userKeyword.value = ''
|
||||
userPage.value = 1
|
||||
if (val) {
|
||||
fetchUserOptions(val, true)
|
||||
}
|
||||
}
|
||||
|
||||
function openUserDropdown() {
|
||||
if (sendForm.tenant_id && userOptions.value.length === 0) {
|
||||
userKeyword.value = ''
|
||||
userPage.value = 1
|
||||
fetchUserOptions(sendForm.tenant_id, true)
|
||||
}
|
||||
}
|
||||
|
||||
// Send message modal
|
||||
const showSendModal = ref(false)
|
||||
const sendLoading = ref(false)
|
||||
const sendForm = reactive({
|
||||
tenant_id: undefined as number | undefined,
|
||||
user_id: '' as string | number,
|
||||
user_id: undefined as number | undefined,
|
||||
title: '',
|
||||
content: '',
|
||||
channel: 'in_app',
|
||||
@@ -136,10 +206,13 @@ const channelOptions = [
|
||||
]
|
||||
|
||||
function openSendModal() {
|
||||
Object.assign(sendForm, { tenant_id: undefined, user_id: '', title: '', content: '', channel: 'in_app' })
|
||||
Object.assign(sendForm, { tenant_id: undefined, user_id: undefined, title: '', content: '', channel: 'in_app' })
|
||||
tenantKeyword.value = ''
|
||||
tenantPage.value = 1
|
||||
tenantOptions.value = []
|
||||
userOptions.value = []
|
||||
userKeyword.value = ''
|
||||
userPage.value = 1
|
||||
showSendModal.value = true
|
||||
}
|
||||
|
||||
@@ -164,7 +237,7 @@ async function handleSendSubmit(done: () => void) {
|
||||
content: sendForm.content,
|
||||
channel: sendForm.channel,
|
||||
}
|
||||
if (sendForm.user_id) payload.user_id = Number(sendForm.user_id)
|
||||
if (sendForm.user_id) payload.user_id = sendForm.user_id
|
||||
await request.post('/admin/notification/messages/send', payload)
|
||||
Message.success('发送成功')
|
||||
done()
|
||||
@@ -271,13 +344,27 @@ onMounted(fetchList)
|
||||
allow-clear
|
||||
:filter-option="false"
|
||||
:fallback-option="true"
|
||||
@change="handleSendTenantChange"
|
||||
@search="handleTenantSearch"
|
||||
@popup-visible-change="(visible: boolean) => { if (visible) openTenantDropdown() }"
|
||||
@scroll-reach-edge="handleTenantScrollReachEdge"
|
||||
/>
|
||||
</AFormItem>
|
||||
<AFormItem label="用户ID">
|
||||
<AInputNumber v-model="sendForm.user_id" :min="1" class="w-full" placeholder="留空则发送给租户所有人" />
|
||||
<AFormItem label="用户" extra="不选则发送给租户所有人">
|
||||
<ASelect
|
||||
v-model="sendForm.user_id"
|
||||
:options="userOptions"
|
||||
:loading="userLoading"
|
||||
:disabled="!sendForm.tenant_id"
|
||||
placeholder="请先选择租户"
|
||||
allow-search
|
||||
allow-clear
|
||||
:filter-option="false"
|
||||
:fallback-option="true"
|
||||
@search="handleUserSearch"
|
||||
@popup-visible-change="(visible: boolean) => { if (visible) openUserDropdown() }"
|
||||
@scroll-reach-edge="handleUserScrollReachEdge"
|
||||
/>
|
||||
</AFormItem>
|
||||
<AFormItem label="标题" required>
|
||||
<AInput v-model="sendForm.title" placeholder="消息标题" />
|
||||
|
||||
@@ -25,7 +25,6 @@ export default defineConfig({
|
||||
'/api': {
|
||||
target: 'http://127.0.0.1:18888',
|
||||
changeOrigin: true,
|
||||
ws: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,11 +26,36 @@ const props = defineProps<{
|
||||
|
||||
const currentIndex = ref(0)
|
||||
const showDetail = ref(false)
|
||||
const dismissedIds = ref<Set<number>>(new Set())
|
||||
const STORAGE_KEY = 'tenant_dismissed_announcements'
|
||||
|
||||
function loadDismissedIds(): Set<number> {
|
||||
try {
|
||||
const raw = localStorage.getItem(STORAGE_KEY)
|
||||
if (raw) return new Set(JSON.parse(raw))
|
||||
} catch { /* ignore */ }
|
||||
return new Set()
|
||||
}
|
||||
|
||||
function saveDismissedIds(ids: Set<number>) {
|
||||
try {
|
||||
localStorage.setItem(STORAGE_KEY, JSON.stringify([...ids]))
|
||||
} catch { /* ignore */ }
|
||||
}
|
||||
|
||||
const dismissedIds = ref<Set<number>>(loadDismissedIds())
|
||||
|
||||
let autoTimer: ReturnType<typeof setInterval> | null = null
|
||||
|
||||
const visibleItems = computed(() => {
|
||||
const activeIds = new Set(props.announcements.map(a => a.id))
|
||||
// 清理已不存在的 dismissed ID
|
||||
const changed = [...dismissedIds.value].some(id => !activeIds.has(id))
|
||||
if (changed) {
|
||||
for (const id of [...dismissedIds.value]) {
|
||||
if (!activeIds.has(id)) dismissedIds.value.delete(id)
|
||||
}
|
||||
saveDismissedIds(dismissedIds.value)
|
||||
}
|
||||
return props.announcements.filter(a => !dismissedIds.value.has(a.id))
|
||||
})
|
||||
|
||||
@@ -54,6 +79,7 @@ function dismissCurrent() {
|
||||
if (!currentItem.value) return
|
||||
if (currentItem.value.type === 'important') return
|
||||
dismissedIds.value.add(currentItem.value.id)
|
||||
saveDismissedIds(dismissedIds.value)
|
||||
if (currentIndex.value >= visibleItems.value.length) {
|
||||
currentIndex.value = 0
|
||||
}
|
||||
|
||||
@@ -2,58 +2,56 @@ import { ref } from 'vue'
|
||||
import request from '@/utils/request'
|
||||
|
||||
const unreadCount = ref(0)
|
||||
let pollTimer: ReturnType<typeof setInterval> | null = null
|
||||
let onNewCallback: ((count: number) => void) | null = null
|
||||
|
||||
async function fetchCount() {
|
||||
try {
|
||||
const res: any = await request.get('/tenant/notifications/unread-count', {
|
||||
_suppressErrorMsg: true,
|
||||
} as any)
|
||||
const newCount = res.data?.data?.unread_count || 0
|
||||
if (newCount > unreadCount.value && onNewCallback) {
|
||||
onNewCallback(newCount - unreadCount.value)
|
||||
}
|
||||
unreadCount.value = newCount
|
||||
} catch {
|
||||
// silently ignore
|
||||
}
|
||||
}
|
||||
|
||||
function startPolling() {
|
||||
fetchCount()
|
||||
if (!pollTimer) {
|
||||
pollTimer = setInterval(fetchCount, 10 * 60 * 1000)
|
||||
}
|
||||
}
|
||||
|
||||
function stopPolling() {
|
||||
if (pollTimer) {
|
||||
clearInterval(pollTimer)
|
||||
pollTimer = null
|
||||
}
|
||||
}
|
||||
|
||||
function decrement() {
|
||||
if (unreadCount.value > 0) unreadCount.value--
|
||||
}
|
||||
|
||||
function reset() {
|
||||
unreadCount.value = 0
|
||||
}
|
||||
|
||||
function setOnNewNotification(cb: (newCount: number) => void) {
|
||||
onNewCallback = cb
|
||||
}
|
||||
|
||||
export function useNotificationCount() {
|
||||
let timer: ReturnType<typeof setInterval> | null = null
|
||||
let previousCount = -1
|
||||
|
||||
async function fetchCount() {
|
||||
try {
|
||||
const res: any = await request.get('/tenant/notifications/unread-count', {
|
||||
_suppressErrorMsg: true,
|
||||
} as any)
|
||||
const newCount = res.data?.data?.count || 0
|
||||
const oldCount = unreadCount.value
|
||||
previousCount = oldCount
|
||||
unreadCount.value = newCount
|
||||
if (previousCount >= 0 && newCount > oldCount && onNewCallback) {
|
||||
onNewCallback(newCount - oldCount)
|
||||
}
|
||||
} catch {
|
||||
// silently ignore
|
||||
}
|
||||
}
|
||||
|
||||
function startPolling(interval = 30_000) {
|
||||
fetchCount()
|
||||
timer = setInterval(fetchCount, interval)
|
||||
}
|
||||
|
||||
function stopPolling() {
|
||||
if (timer) {
|
||||
clearInterval(timer)
|
||||
timer = null
|
||||
}
|
||||
}
|
||||
|
||||
function decrement() {
|
||||
if (unreadCount.value > 0) unreadCount.value--
|
||||
}
|
||||
|
||||
function reset() {
|
||||
unreadCount.value = 0
|
||||
}
|
||||
|
||||
function setOnNewNotification(cb: (newCount: number) => void) {
|
||||
onNewCallback = cb
|
||||
}
|
||||
|
||||
return {
|
||||
unreadCount,
|
||||
fetchCount,
|
||||
startPolling,
|
||||
stopPolling,
|
||||
fetchCount,
|
||||
decrement,
|
||||
reset,
|
||||
setOnNewNotification,
|
||||
|
||||
@@ -96,6 +96,11 @@ function closeUserMenu() {
|
||||
}
|
||||
|
||||
async function handleLogout() {
|
||||
stopNotificationPolling()
|
||||
if (announcementTimer) {
|
||||
clearInterval(announcementTimer)
|
||||
announcementTimer = null
|
||||
}
|
||||
await authStore.logout()
|
||||
router.push('/tenant/login')
|
||||
}
|
||||
@@ -128,8 +133,8 @@ onMounted(async () => {
|
||||
authStore.loadFromStorage()
|
||||
document.addEventListener('click', handleClickOutside)
|
||||
fetchAnnouncements()
|
||||
announcementTimer = setInterval(fetchAnnouncements, 60_000)
|
||||
startNotificationPolling(30_000)
|
||||
announcementTimer = setInterval(fetchAnnouncements, 10 * 60 * 1000)
|
||||
startNotificationPolling()
|
||||
|
||||
await fetchPublicSettings()
|
||||
if (publicSettings.demo_mode) {
|
||||
@@ -140,8 +145,11 @@ onMounted(async () => {
|
||||
|
||||
onBeforeUnmount(() => {
|
||||
document.removeEventListener('click', handleClickOutside)
|
||||
if (announcementTimer) clearInterval(announcementTimer)
|
||||
stopNotificationPolling()
|
||||
if (announcementTimer) {
|
||||
clearInterval(announcementTimer)
|
||||
announcementTimer = null
|
||||
}
|
||||
unmountWatermark()
|
||||
})
|
||||
</script>
|
||||
|
||||
@@ -11,6 +11,7 @@ interface Notification {
|
||||
content: string
|
||||
type: string
|
||||
is_read: boolean
|
||||
is_broadcast: number
|
||||
created_at: string
|
||||
}
|
||||
|
||||
@@ -92,6 +93,19 @@ async function markAllRead() {
|
||||
}
|
||||
}
|
||||
|
||||
const confirmDeleteId = ref<number | null>(null)
|
||||
|
||||
async function deleteNotification(n: Notification) {
|
||||
try {
|
||||
await request.delete(`/tenant/notifications/${n.id}`)
|
||||
notifications.value = notifications.value.filter((item) => item.id !== n.id)
|
||||
total.value--
|
||||
expandedId.value = null
|
||||
confirmDeleteId.value = null
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
function toggleNotification(n: Notification) {
|
||||
if (expandedId.value === n.id) {
|
||||
expandedId.value = null
|
||||
@@ -219,6 +233,18 @@ onMounted(() => {
|
||||
<!-- Expanded content -->
|
||||
<div v-if="expandedId === n.id" class="mt-3 pt-3 border-t border-gray-100">
|
||||
<p class="text-sm text-gray-700 whitespace-pre-wrap leading-relaxed">{{ n.content }}</p>
|
||||
<!-- Delete button for read personal messages -->
|
||||
<div v-if="n.is_read && !n.is_broadcast" class="mt-3 flex items-center justify-end gap-2">
|
||||
<template v-if="confirmDeleteId === n.id">
|
||||
<span class="text-xs text-gray-500">确认删除？</span>
|
||||
<button class="btn btn-danger btn-sm" @click.stop="deleteNotification(n)">删除</button>
|
||||
<button class="btn btn-secondary btn-sm" @click.stop="confirmDeleteId = null">取消</button>
|
||||
</template>
|
||||
<button v-else class="btn btn-ghost btn-sm text-red-500 hover:bg-red-50" @click.stop="confirmDeleteId = n.id">
|
||||
<svg class="h-4 w-4" fill="none" viewBox="0 0 24 24" stroke="currentColor" stroke-width="1.5"><path stroke-linecap="round" stroke-linejoin="round" d="M14.74 9l-.346 9m-4.788 0L9.26 9m9.968-3.21c.342.052.682.107 1.022.166m-1.022-.165L18.16 19.673a2.25 2.25 0 01-2.244 2.077H8.084a2.25 2.25 0 01-2.244-2.077L4.772 5.79m14.456 0a48.108 48.108 0 00-3.478-.397m-12 .562c.34-.059.68-.114 1.022-.165m0 0a48.11 48.11 0 013.478-.397m7.5 0v-.916c0-1.18-.91-2.164-2.09-2.201a51.964 51.964 0 00-3.32 0c-1.18.037-2.09 1.022-2.09 2.201v.916m7.5 0a48.667 48.667 0 00-7.5 0"/></svg>
|
||||
删除
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
Reference in new issue
Block a user