kefu_cloud/server/internal/ws/bus.go

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()
	}
}