From 10d2e2ed3d3a08d6c2e9ac8d3294c9e8e501603f Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sat, 5 Sep 2026 14:39:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20WebSocket=20=E5=B9=BF=E6=92=AD?= =?UTF-8?q?=E6=94=AF=E6=8C=81=20Redis=20Pub/Sub=20=E8=B7=A8=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E6=89=87=E5=87=BA=EF=BC=88REDIS=5FADDR=20=E5=8F=AF?= =?UTF-8?q?=E9=80=89=E5=BC=80=E5=85=B3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 ws.EventBus 接口与 RedisBus 实现；Hub 五类广播改为发布信封、各节点订阅后本地投递 - 单实例不设 REDIS_ADDR 保持进程内直投；Redis 不可用时自动兜底本地投递 - config 新增 REDIS_ADDR/REDIS_PASSWORD/REDIS_DB；dev compose 增加 --profile redis (6380) - prod compose 与 .env.prod.example 增加多副本配置说明；README 更新已知边界 - 已用双实例 + 真实 Redis 实测跨节点推送；新增总线单测（含发布失败兜底），-race 通过 --- README.md | 10 +- deploy/.env.prod.example | 6 ++ docker-compose.dev.yml | 13 +++ docker-compose.prod.yml | 16 +++ server/cmd/main.go | 7 ++ server/go.mod | 2 + server/go.sum | 8 ++ server/internal/config/config.go | 13 +++ server/internal/ws/bus.go | 170 +++++++++++++++++++++++++++++++ server/internal/ws/bus_test.go | 152 +++++++++++++++++++++++++++ server/internal/ws/ws.go | 39 ++++++- 11 files changed, 430 insertions(+), 6 deletions(-) create mode 100644 server/internal/ws/bus.go create mode 100644 server/internal/ws/bus_test.go diff --git a/README.md b/README.md index 3215159..4c74ada 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ | 后端 | Go、Gin、GORM、JWT、Gorilla WebSocket | | 数据库 | PostgreSQL（生产/开发主库）；测试可用 SQLite | | 存储 | MinIO（S3 兼容），图片转 WebP | -| 实时 | 进程内 WebSocket Hub（单实例；多副本需后续 Redis 广播） | +| 实时 | 进程内 WebSocket Hub；多 API 副本时通过 `REDIS_ADDR` 启用 Redis Pub/Sub 跨节点广播 | ### 消息链路（简要） @@ -211,6 +211,8 @@ make db-up | `STORAGE_ACCESS_KEY` / `STORAGE_SECRET_KEY` | `minioadmin` | | | `STORAGE_BUCKET` | `kefu` | | | `STORAGE_PUBLIC_BASE_URL` | `http://localhost:9100/kefu` | 浏览器访问前缀 | +| `REDIS_ADDR` | 空 | 为空=单实例进程内广播；设置后 WS 广播走 Redis Pub/Sub 跨节点扇出 | +| `REDIS_PASSWORD` / `REDIS_DB` | 空 / `0` | 与 `REDIS_ADDR` 配合使用 | --- @@ -270,8 +272,8 @@ cd web && npx tsc --noEmit | 项 | 说明 | |----|------| -| 多 API 实例 | Hub 为进程内存；扩副本需 Redis Pub/Sub 等跨节点广播 | -| 独立消息队列 | 当前不需要；实时靠 WS Hub | +| 多 API 实例 | 设置 `REDIS_ADDR` 后 Hub 广播经 Redis Pub/Sub 跨节点扇出（各节点本地投递、连接仍节点本地，需入口层 sticky/共享连接）；单实例无需 Redis | +| 独立消息队列 | 当前不需要；实时靠 WS Hub（+ 可选 Redis 广播） | | 技能组 / 渠道专属坐席 | 未做；可在候选过滤层扩展 | | 厂商推送（APNs/FCM） | 未做 | | 微信/APP/电话渠道 | 模型预留，完整对接未做 | @@ -279,7 +281,7 @@ cd web && npx tsc --noEmit ### 建议演进顺序 1. 生产配置加固（JWT、HTTPS、存储与 DB 备份） -2. 需要水平扩展时：Redis 会话广播 + sticky 或共享连接层 +2. 需要水平扩展时：`REDIS_ADDR` 启用跨节点广播 + sticky 或共享连接层（已内置） 3. 技能组 / 渠道路由 4. 监控可对接 Prometheus（现有 metrics 可作过渡） diff --git a/deploy/.env.prod.example b/deploy/.env.prod.example index a12eeb1..7cc6810 100644 --- a/deploy/.env.prod.example +++ b/deploy/.env.prod.example @@ -39,6 +39,12 @@ STORAGE_WEBP_QUALITY=80 GIN_MODE=release IMAGE_TAG=latest +# ========== 跨节点 WebSocket 广播（可选）========== +# API 扩多副本时启用（需为所有 api 副本配置相同值）；单实例留空即进程内广播 +# REDIS_ADDR=redis:6379 +# REDIS_PASSWORD= +# REDIS_DB=0 + # 若使用腾讯云 COS 替代 MinIO，可去掉 compose 中 minio 服务，并设置： # STORAGE_ENDPOINT=cos.ap-guangzhou.myqcloud.com # STORAGE_USE_SSL=true diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 498adf0..e176c7c 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -42,6 +42,19 @@ services: echo 'MinIO bucket kefu ready'; " + # 可选：跨节点 WebSocket 广播总线（多 API 副本联调时启用） + # docker compose -f docker-compose.dev.yml --profile redis up -d + # 后端再设置 REDIS_ADDR=localhost:6380 + redis: + image: redis:7-alpine + container_name: kefu-cloud-redis + profiles: ["redis"] + ports: + - "6380:6379" + volumes: + - redisdata:/data + volumes: pgdata: miniodata: + redisdata: diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index ca0f3e6..0838cbc 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -66,6 +66,17 @@ services: MINIO_ROOT_PASSWORD: ${STORAGE_SECRET_KEY} STORAGE_BUCKET: ${STORAGE_BUCKET:-kefu} + # 可选：跨节点 WebSocket 广播总线（API 扩多副本时启用） + # .env.prod 设置 REDIS_ADDR=redis:6379 并为 api 副本配置相同值 + redis: + image: redis:7-alpine + container_name: kefu-cloud-redis + restart: unless-stopped + volumes: + - kefu_cloud_redisdata:/data + networks: + - kefu_cloud_net + api: build: context: . @@ -100,6 +111,10 @@ services: STORAGE_MAX_UPLOAD_MB: ${STORAGE_MAX_UPLOAD_MB:-10} STORAGE_MAX_IMAGE_EDGE: ${STORAGE_MAX_IMAGE_EDGE:-1920} STORAGE_WEBP_QUALITY: ${STORAGE_WEBP_QUALITY:-80} + # 跨节点 WebSocket 广播（多副本时设为 redis:6379；单实例留空即进程内广播） + REDIS_ADDR: ${REDIS_ADDR:-} + REDIS_PASSWORD: ${REDIS_PASSWORD:-} + REDIS_DB: ${REDIS_DB:-0} networks: - kefu_cloud_net # 不对外暴露 8080，统一走 nginx @@ -158,3 +173,4 @@ networks: volumes: kefu_cloud_pgdata: kefu_cloud_miniodata: + kefu_cloud_redisdata: diff --git a/server/cmd/main.go b/server/cmd/main.go index ad0e034..aafc99d 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -40,6 +40,13 @@ func main() { // 启动 WebSocket Hub ws.SetAllowedOrigins(cfg.AllowedOrigins) + // 多 API 副本部署时通过 REDIS_ADDR 启用跨节点广播；单实例保持进程内投递 + if cfg.Redis.Addr != "" { + bus := ws.NewRedisBus(cfg.Redis.Addr, cfg.Redis.Password, cfg.Redis.DB) + ws.DefaultHub.SetEventBus(bus) + log.Printf("WebSocket 跨节点广播已启用: redis=%s db=%d", cfg.Redis.Addr, cfg.Redis.DB) + defer func() { _ = bus.Close() }() + } go ws.DefaultHub.Run() // 设置 Gin 模式 diff --git a/server/go.mod b/server/go.mod index dba0885..94826b4 100644 --- a/server/go.mod +++ b/server/go.mod @@ -9,6 +9,7 @@ require ( github.com/golang-jwt/jwt/v5 v5.2.1 github.com/gorilla/websocket v1.5.3 github.com/minio/minio-go/v7 v7.2.1 + github.com/redis/go-redis/v9 v9.7.0 github.com/shirou/gopsutil/v4 v4.25.1 golang.org/x/crypto v0.51.0 gorm.io/driver/postgres v1.5.9 @@ -22,6 +23,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.4 // indirect github.com/cloudwego/iasm v0.2.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/ebitengine/purego v0.8.2 // indirect github.com/gabriel-vasile/mimetype v1.4.3 // indirect diff --git a/server/go.sum b/server/go.sum index 1a444d7..97e3ff8 100644 --- a/server/go.sum +++ b/server/go.sum @@ -1,3 +1,7 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc0= github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4= github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM= @@ -14,6 +18,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/disintegration/imaging v1.6.2 h1:w1LecBlG2Lnp8B3jk5zSuNqd7b4DXhcjwek1ei82L+c= github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= @@ -102,6 +108,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= +github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= github.com/rogpeppe/go-internal v1.15.0 h1:D0RCU5rMAp+SpgkiNdrjfJ+LX4J1M32V2NeCY7EJ6hc= github.com/rogpeppe/go-internal v1.15.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs= github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 84c8d26..73e415e 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -12,6 +12,7 @@ type Config struct { Database DatabaseConfig JWT JWTConfig Storage StorageConfig + Redis RedisConfig AllowedOrigins []string } @@ -48,6 +49,13 @@ type StorageConfig struct { WebPQuality float32 } +// RedisConfig 可选的跨节点 WebSocket 广播总线（多 API 副本部署时启用）。 +type RedisConfig struct { + Addr string // 为空表示单实例模式，WS 广播仅进程内 + Password string + DB int +} + func Load() *Config { return &Config{ Server: ServerConfig{ @@ -80,6 +88,11 @@ func Load() *Config { WebPQuality: float32(getEnvInt("STORAGE_WEBP_QUALITY", 80)), }, AllowedOrigins: splitCSV(getEnv("CORS_ALLOW_ORIGINS", "*")), + Redis: RedisConfig{ + Addr: getEnv("REDIS_ADDR", ""), + Password: getEnv("REDIS_PASSWORD", ""), + DB: getEnvInt("REDIS_DB", 0), + }, } } diff --git a/server/internal/ws/bus.go b/server/internal/ws/bus.go new file mode 100644 index 0000000..25673ed --- /dev/null +++ b/server/internal/ws/bus.go @@ -0,0 +1,170 @@ +package ws + +import ( + "context" + "encoding/json" + "log" + "os" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +// busScope 广播信封的路由范围。 +const ( + busScopeSession = "session" // 会话相关（坐席 + 该会话访客） + busScopeTenantStaff = "tenant_staff" // 租户内工作人员 + busScopeVisitor = "visitor" // 指定会话访客 + busScopeSessionStaff = "session_staff" // 可查看会话的工作人员（不含访客） + busScopeDisconnect = "disconnect" // 断开指定用户所有连接 +) + +// busEnvelope 跨节点广播信封：scope + 路由参数 + 原始 WS 消息。 +type busEnvelope struct { + Node string `json:"node,omitempty"` + Scope string `json:"scope"` + TenantID uint `json:"tenant_id"` + SessionID uint `json:"session_id,omitempty"` + AgentID *uint `json:"agent_id,omitempty"` + UserID uint `json:"user_id,omitempty"` + Text string `json:"text,omitempty"` + Message json.RawMessage `json:"message,omitempty"` +} + +// EventBus 跨节点广播总线（Redis Pub/Sub 实现）。 +// 单实例部署无需设置；多副本部署通过 REDIS_ADDR 启用后， +// Hub 的所有广播改为发布到总线，由各节点订阅后在本进程内投递。 +type EventBus interface { + Publish(env *busEnvelope) error + Subscribe(dispatch func(env *busEnvelope)) + Close() error +} + +// RedisBus 基于 Redis Pub/Sub 的 EventBus。 +type RedisBus struct { + client *redis.Client + channel string + nodeID string + ctx context.Context + cancel context.CancelFunc + closeMu sync.Once +} + +// NewRedisBus 创建 Redis 广播总线（不立即建连，首次命令时惰性连接）。 +func NewRedisBus(addr, password string, db int) *RedisBus { + ctx, cancel := context.WithCancel(context.Background()) + nodeID, _ := os.Hostname() + return &RedisBus{ + client: redis.NewClient(&redis.Options{ + Addr: addr, + Password: password, + DB: db, + }), + channel: "kefu:ws:broadcast", + nodeID: nodeID, + ctx: ctx, + cancel: cancel, + } +} + +// Publish 发布信封；Redis 不可用时返回错误，由调用方决定本地兜底。 +func (b *RedisBus) Publish(env *busEnvelope) error { + if env.Node == "" { + env.Node = b.nodeID + } + payload, err := json.Marshal(env) + if err != nil { + return err + } + return b.client.Publish(b.ctx, b.channel, payload).Err() +} + +// Subscribe 阻塞订阅总线，收到信封后回调 dispatch（含本节点发出的消息， +// 由 dispatch 统一本进程投递，避免双发）。 +func (b *RedisBus) Subscribe(dispatch func(env *busEnvelope)) { + for { + select { + case <-b.ctx.Done(): + return + default: + } + sub := b.client.Subscribe(b.ctx, b.channel) + // 等待订阅确认失败（Redis 未就绪）时退避重试 + if _, err := sub.Receive(b.ctx); err != nil { + _ = sub.Close() + select { + case <-b.ctx.Done(): + return + case <-time.After(2 * time.Second): + continue + } + } + ch := sub.Channel() + for msg := range ch { + var env busEnvelope + if json.Unmarshal([]byte(msg.Payload), &env) != nil || env.Scope == "" { + continue + } + dispatch(&env) + } + _ = sub.Close() + // Channel 关闭（连接断开）后重建订阅 + select { + case <-b.ctx.Done(): + return + case <-time.After(time.Second): + } + } +} + +// Close 释放订阅与连接。 +func (b *RedisBus) Close() error { + var err error + b.closeMu.Do(func() { + b.cancel() + err = b.client.Close() + }) + return err +} + +// SetEventBus 挂载跨节点广播总线（须在 Hub.Run 与首个连接建立前调用）。 +func (h *Hub) SetEventBus(bus EventBus) { + h.mu.Lock() + h.bus = bus + h.mu.Unlock() + go bus.Subscribe(h.dispatch) +} + +// dispatch 总线信封 → 本进程投递。 +func (h *Hub) dispatch(env *busEnvelope) { + switch env.Scope { + case busScopeSession: + h.deliverToSession(env.TenantID, env.SessionID, env.AgentID, env.Message) + case busScopeTenantStaff: + h.deliverToTenantStaff(env.TenantID, env.Message) + case busScopeVisitor: + h.deliverToVisitor(env.TenantID, env.SessionID, env.Message) + case busScopeSessionStaff: + h.deliverToSessionStaff(env.TenantID, env.AgentID, env.Message) + case busScopeDisconnect: + h.disconnectUserLocal(env.UserID, env.Text) + default: + log.Printf("ws: 未知广播 scope=%q", env.Scope) + } +} + +// publishOrDeliver 优先走总线；总线不可用时兜底本进程投递，保证单点功能不回退。 +func (h *Hub) publishOrDeliver(env *busEnvelope, deliver func()) { + h.mu.RLock() + bus := h.bus + h.mu.RUnlock() + if bus == nil { + deliver() + return + } + if err := bus.Publish(env); err != nil { + log.Printf("ws: 广播发布失败（scope=%s），本进程兜底: %v", env.Scope, err) + deliver() + } +} diff --git a/server/internal/ws/bus_test.go b/server/internal/ws/bus_test.go new file mode 100644 index 0000000..58c54d1 --- /dev/null +++ b/server/internal/ws/bus_test.go @@ -0,0 +1,152 @@ +package ws + +import ( + "errors" + "sync" + "testing" + "time" +) + +type fakeBus struct { + mu sync.Mutex + published []*busEnvelope + fail bool +} + +func (f *fakeBus) Publish(env *busEnvelope) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.fail { + return errors.New("redis down") + } + f.published = append(f.published, env) + return nil +} + +func (f *fakeBus) Subscribe(dispatch func(env *busEnvelope)) {} + +func (f *fakeBus) Close() error { return nil } + +func (f *fakeBus) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.published) +} + +func TestBroadcastPublishesToBusInsteadOfLocalDelivery(t *testing.T) { + hub := NewHub() + go hub.Run() + bus := &fakeBus{} + hub.mu.Lock() + hub.bus = bus + hub.mu.Unlock() + + sessionID := uint(12) + agentID := uint(7) + assigned := &Client{TenantID: 1, UserID: agentID, Role: "agent", Kind: "agent", Send: make(chan []byte, 1)} + visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &sessionID, Send: make(chan []byte, 1)} + for _, client := range []*Client{assigned, visitor} { + hub.register <- client + } + + hub.BroadcastToSession(1, sessionID, &agentID, []byte(`{"type":"message"}`)) + if bus.count() != 1 { + t.Fatalf("应向总线发布 1 条信封，got=%d", bus.count()) + } + // 发布模式下本进程不直接投递（由订阅回调统一投递，避免双发） + select { + case <-assigned.Send: + t.Fatal("发布模式不应直接投递本地客户端") + case <-time.After(100 * time.Millisecond): + } + + // 订阅回调 dispatch 后应在本进程按范围投递 + hub.dispatch(&busEnvelope{Scope: busScopeSession, TenantID: 1, SessionID: sessionID, AgentID: &agentID, Message: []byte(`{"type":"message"}`)}) + for _, client := range []*Client{assigned, visitor} { + select { + case <-client.Send: + case <-time.After(time.Second): + t.Fatalf("dispatch 后客户端未收到消息：%+v", client) + } + } +} + +func TestDispatchRoutesAllScopes(t *testing.T) { + hub := NewHub() + go hub.Run() + + sessionID := uint(41) + agentID := uint(3) + assigned := &Client{TenantID: 1, UserID: agentID, Role: "agent", Kind: "agent", Send: make(chan []byte, 1)} + supervisor := &Client{TenantID: 1, UserID: 4, Role: "supervisor", Kind: "agent", Send: make(chan []byte, 1)} + visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &sessionID, Send: make(chan []byte, 1)} + other := &Client{TenantID: 2, Kind: "agent", UserID: 9, Role: "agent", Send: make(chan []byte, 1)} + for _, client := range []*Client{assigned, supervisor, visitor, other} { + hub.register <- client + } + + // tenant_staff + hub.dispatch(&busEnvelope{Scope: busScopeTenantStaff, TenantID: 1, Message: []byte(`{"type":"session_updated","session_id":41}`)}) + for _, client := range []*Client{assigned, supervisor} { + select { + case <-client.Send: + case <-time.After(time.Second): + t.Fatalf("租户工作人员应收到 session_updated：%+v", client) + } + } + select { + case <-visitor.Send: + t.Fatal("访客不应收到 tenant_staff 广播") + case <-other.Send: + t.Fatal("跨租户客户端不应收到广播") + case <-time.After(100 * time.Millisecond): + } + + // visitor + hub.dispatch(&busEnvelope{Scope: busScopeVisitor, TenantID: 1, SessionID: sessionID, Message: []byte(`{"type":"typing"}`)}) + select { + case <-visitor.Send: + case <-time.After(time.Second): + t.Fatal("会话访客应收到消息") + } + + // disconnect：跨节点踢下线（踢 supervisor，避免 assigned/visitor 被断开） + hub.dispatch(&busEnvelope{Scope: busScopeDisconnect, UserID: 4, Text: "账号已停用"}) + deadline := time.After(time.Second) + for { + select { + case <-supervisor.Send: + // 先收到 kicked 通知，继续等连接关闭 + case <-deadline: + t.Fatal("被踢客户端未收到任何消息") + default: + } + hub.mu.RLock() + _, stillOnline := hub.clients[supervisor] + hub.mu.RUnlock() + if !stillOnline { + return + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestPublishFailureFallsBackToLocalDelivery(t *testing.T) { + hub := NewHub() + go hub.Run() + bus := &fakeBus{fail: true} + hub.mu.Lock() + hub.bus = bus + hub.mu.Unlock() + + visitorSession := uint(51) + visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &visitorSession, Send: make(chan []byte, 1)} + hub.register <- visitor + + hub.BroadcastToVisitor(1, visitorSession, []byte(`{"type":"typing"}`)) + select { + case <-visitor.Send: + case <-time.After(time.Second): + t.Fatal("总线不可用时应兜底本进程投递") + } +} diff --git a/server/internal/ws/ws.go b/server/internal/ws/ws.go index fe8ffbd..9e7ce1c 100644 --- a/server/internal/ws/ws.go +++ b/server/internal/ws/ws.go @@ -87,6 +87,8 @@ type Hub struct { register chan *Client unregister chan *Client mu sync.RWMutex + // bus 跨节点广播总线；nil 表示单实例模式（进程内直接投递） + bus EventBus } var DefaultHub = NewHub() @@ -127,8 +129,14 @@ func (h *Hub) Run() { } } -// DisconnectUser 断开指定用户的所有 WebSocket 连接并推送通知。 +// DisconnectUser 断开指定用户在所有节点上的 WebSocket 连接并推送通知。 func (h *Hub) DisconnectUser(userID uint, message string) { + h.publishOrDeliver(&busEnvelope{Scope: busScopeDisconnect, UserID: userID, Text: message}, func() { + h.disconnectUserLocal(userID, message) + }) +} + +func (h *Hub) disconnectUserLocal(userID uint, message string) { h.mu.Lock() defer h.mu.Unlock() for client := range h.clients { @@ -235,6 +243,14 @@ func (h *Hub) send(client *Client, message []byte) { // BroadcastToSession 仅把消息推送给当前客服、主管/管理员及该会话的访客。 func (h *Hub) BroadcastToSession(tenantID, sessionID uint, agentID *uint, message []byte) { + h.publishOrDeliver(&busEnvelope{ + Scope: busScopeSession, TenantID: tenantID, SessionID: sessionID, AgentID: agentID, Message: message, + }, func() { + h.deliverToSession(tenantID, sessionID, agentID, message) + }) +} + +func (h *Hub) deliverToSession(tenantID, sessionID uint, agentID *uint, message []byte) { h.mu.RLock() defer h.mu.RUnlock() @@ -258,13 +274,20 @@ func (h *Hub) BroadcastToSession(tenantID, sessionID uint, agentID *uint, messag // BroadcastToTenantStaff 只通知租户内的工作人员，不向访客泄露其他会话事件。 func (h *Hub) BroadcastToTenantStaff(tenantID uint, message []byte) { + h.publishOrDeliver(&busEnvelope{Scope: busScopeTenantStaff, TenantID: tenantID, Message: message}, func() { + h.deliverToTenantStaff(tenantID, message) + }) +} + +func (h *Hub) deliverToTenantStaff(tenantID uint, message []byte) { h.mu.RLock() defer h.mu.RUnlock() var event Event _ = json.Unmarshal(message, &event) var session *model.Session - if event.SessionID != 0 { + // 单元测试未初始化数据库时跳过会话查询（与 canReceiveSessionEvent 的处理一致） + if model.DB != nil && event.SessionID != 0 { var current model.Session if err := model.DB.Where("id = ? AND tenant_id = ?", event.SessionID, tenantID).First(¤t).Error; err == nil { session = ¤t @@ -280,6 +303,12 @@ func (h *Hub) BroadcastToTenantStaff(tenantID uint, message []byte) { } func (h *Hub) BroadcastToVisitor(tenantID, sessionID uint, message []byte) { + h.publishOrDeliver(&busEnvelope{Scope: busScopeVisitor, TenantID: tenantID, SessionID: sessionID, Message: message}, func() { + h.deliverToVisitor(tenantID, sessionID, message) + }) +} + +func (h *Hub) deliverToVisitor(tenantID, sessionID uint, message []byte) { h.mu.RLock() defer h.mu.RUnlock() @@ -292,6 +321,12 @@ func (h *Hub) BroadcastToVisitor(tenantID, sessionID uint, message []byte) { // BroadcastToSessionStaff 仅推送给可查看该会话的工作人员（不含访客）。 func (h *Hub) BroadcastToSessionStaff(tenantID uint, agentID *uint, message []byte) { + h.publishOrDeliver(&busEnvelope{Scope: busScopeSessionStaff, TenantID: tenantID, AgentID: agentID, Message: message}, func() { + h.deliverToSessionStaff(tenantID, agentID, message) + }) +} + +func (h *Hub) deliverToSessionStaff(tenantID uint, agentID *uint, message []byte) { h.mu.RLock() defer h.mu.RUnlock()