yml2213
d105840b1f
- 工作台会话列表支持状态/紧急筛选 - 访客 Widget 支持 jpg/png/gif 图片预览发送 - WebSocket 支持访客输入中通知客服，客服输入中通知访客 - 工作台聊天区展示「访客正在输入」动画
279 lines
6.7 KiB
Go
279 lines
6.7 KiB
Go
package ws
|
|
|
|
import (
|
|
"encoding/json"
|
|
"log"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/gorilla/websocket"
|
|
"kefu-sys/server/internal/model"
|
|
)
|
|
|
|
var upgrader = websocket.Upgrader{
|
|
CheckOrigin: func(r *http.Request) bool { return true },
|
|
Subprotocols: []string{"kefu-v1", "kefu-visitor-v1"},
|
|
}
|
|
|
|
type Client struct {
|
|
Conn *websocket.Conn
|
|
UserID uint
|
|
TenantID uint
|
|
Role string
|
|
Kind string
|
|
SessionID *uint
|
|
Send chan []byte
|
|
}
|
|
|
|
type Event struct {
|
|
Type string `json:"type"`
|
|
SessionID uint `json:"session_id,omitempty"`
|
|
Data interface{} `json:"data,omitempty"`
|
|
Timestamp int64 `json:"timestamp"`
|
|
}
|
|
|
|
type ClientEvent struct {
|
|
Type string `json:"type"`
|
|
SessionID uint `json:"session_id"`
|
|
}
|
|
|
|
type Hub struct {
|
|
clients map[*Client]bool
|
|
register chan *Client
|
|
unregister chan *Client
|
|
mu sync.RWMutex
|
|
}
|
|
|
|
var DefaultHub = NewHub()
|
|
|
|
func NewHub() *Hub {
|
|
return &Hub{
|
|
clients: make(map[*Client]bool),
|
|
register: make(chan *Client),
|
|
unregister: make(chan *Client),
|
|
}
|
|
}
|
|
|
|
func NewEvent(eventType string, sessionID uint, data interface{}) ([]byte, error) {
|
|
return json.Marshal(Event{
|
|
Type: eventType,
|
|
SessionID: sessionID,
|
|
Data: data,
|
|
Timestamp: time.Now().UnixMilli(),
|
|
})
|
|
}
|
|
|
|
func (h *Hub) Run() {
|
|
for {
|
|
select {
|
|
case client := <-h.register:
|
|
h.mu.Lock()
|
|
h.clients[client] = true
|
|
h.mu.Unlock()
|
|
|
|
case client := <-h.unregister:
|
|
h.mu.Lock()
|
|
if _, ok := h.clients[client]; ok {
|
|
delete(h.clients, client)
|
|
close(client.Send)
|
|
}
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *Hub) send(client *Client, message []byte) {
|
|
select {
|
|
case client.Send <- message:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// BroadcastToSession 仅把消息推送给当前客服、主管/管理员及该会话的访客。
|
|
func (h *Hub) BroadcastToSession(tenantID, sessionID uint, agentID *uint, message []byte) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
for client := range h.clients {
|
|
if client.TenantID != tenantID {
|
|
continue
|
|
}
|
|
if client.Kind == "visitor" {
|
|
if client.SessionID != nil && *client.SessionID == sessionID {
|
|
h.send(client, message)
|
|
}
|
|
continue
|
|
}
|
|
if client.Role == "admin" || client.Role == "supervisor" ||
|
|
(agentID != nil && client.Role == "agent" && client.UserID == *agentID) {
|
|
h.send(client, message)
|
|
}
|
|
}
|
|
}
|
|
|
|
// BroadcastToTenantStaff 只通知租户内的工作人员，不向访客泄露其他会话事件。
|
|
func (h *Hub) BroadcastToTenantStaff(tenantID uint, message []byte) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
for client := range h.clients {
|
|
if client.TenantID == tenantID && client.Kind == "agent" {
|
|
h.send(client, message)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *Hub) BroadcastToVisitor(tenantID, sessionID uint, message []byte) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
for client := range h.clients {
|
|
if client.TenantID == tenantID && client.Kind == "visitor" && client.SessionID != nil && *client.SessionID == sessionID {
|
|
h.send(client, message)
|
|
}
|
|
}
|
|
}
|
|
|
|
// BroadcastToSessionStaff 仅推送给可查看该会话的工作人员（不含访客）。
|
|
func (h *Hub) BroadcastToSessionStaff(tenantID uint, agentID *uint, message []byte) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
for client := range h.clients {
|
|
if client.TenantID != tenantID || client.Kind != "agent" {
|
|
continue
|
|
}
|
|
if client.Role == "admin" || client.Role == "supervisor" ||
|
|
(agentID != nil && client.Role == "agent" && client.UserID == *agentID) {
|
|
h.send(client, message)
|
|
}
|
|
}
|
|
}
|
|
|
|
func handleClientEvent(client *Client, event ClientEvent) {
|
|
if event.Type != "typing" || event.SessionID == 0 {
|
|
return
|
|
}
|
|
var session model.Session
|
|
if err := model.DB.Where("id = ? AND tenant_id = ?", event.SessionID, client.TenantID).First(&session).Error; err != nil {
|
|
return
|
|
}
|
|
|
|
// 访客输入中 → 通知可接待的客服
|
|
if client.Kind == "visitor" {
|
|
if client.SessionID == nil || *client.SessionID != event.SessionID {
|
|
return
|
|
}
|
|
if session.Status == "ended" || session.Status == "archived" {
|
|
return
|
|
}
|
|
payload, err := NewEvent("typing", session.ID, map[string]string{"from": "visitor"})
|
|
if err != nil {
|
|
return
|
|
}
|
|
if session.AgentID == nil {
|
|
DefaultHub.BroadcastToTenantStaff(session.TenantID, payload)
|
|
return
|
|
}
|
|
DefaultHub.BroadcastToSessionStaff(session.TenantID, session.AgentID, payload)
|
|
return
|
|
}
|
|
|
|
// 客服输入中 → 通知访客
|
|
if client.Kind != "agent" {
|
|
return
|
|
}
|
|
if client.Role == "agent" && (session.AgentID == nil || *session.AgentID != client.UserID) {
|
|
return
|
|
}
|
|
if client.Role != "agent" && client.Role != "admin" && client.Role != "supervisor" {
|
|
return
|
|
}
|
|
payload, err := NewEvent("typing", session.ID, map[string]string{"from": "agent"})
|
|
if err == nil {
|
|
DefaultHub.BroadcastToVisitor(session.TenantID, session.ID, payload)
|
|
}
|
|
}
|
|
|
|
func HandleWebSocket(client *Client) {
|
|
defer func() {
|
|
DefaultHub.unregister <- client
|
|
client.Conn.Close()
|
|
}()
|
|
|
|
client.Conn.SetReadLimit(1024)
|
|
client.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
|
client.Conn.SetPongHandler(func(string) error {
|
|
client.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
|
return nil
|
|
})
|
|
go writePump(client)
|
|
for {
|
|
_, message, err := client.Conn.ReadMessage()
|
|
if err != nil {
|
|
return
|
|
}
|
|
var event ClientEvent
|
|
if json.Unmarshal(message, &event) == nil {
|
|
handleClientEvent(client, event)
|
|
}
|
|
}
|
|
}
|
|
|
|
func writePump(client *Client) {
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case message, ok := <-client.Send:
|
|
if !ok {
|
|
client.Conn.WriteMessage(websocket.CloseMessage, []byte{})
|
|
return
|
|
}
|
|
client.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
|
if err := client.Conn.WriteMessage(websocket.TextMessage, message); err != nil {
|
|
return
|
|
}
|
|
case <-ticker.C:
|
|
client.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
|
if err := client.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func upgrade(w http.ResponseWriter, r *http.Request, client *Client) (*Client, error) {
|
|
conn, err := upgrader.Upgrade(w, r, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
client.Conn = conn
|
|
DefaultHub.register <- client
|
|
return client, nil
|
|
}
|
|
|
|
func UpgradeAgent(w http.ResponseWriter, r *http.Request, userID, tenantID uint, role string) (*Client, error) {
|
|
client := &Client{
|
|
UserID: userID,
|
|
TenantID: tenantID,
|
|
Role: role,
|
|
Kind: "agent",
|
|
Send: make(chan []byte, 256),
|
|
}
|
|
log.Printf("WebSocket 连接: user=%d tenant=%d", userID, tenantID)
|
|
return upgrade(w, r, client)
|
|
}
|
|
|
|
func UpgradeVisitor(w http.ResponseWriter, r *http.Request, tenantID, sessionID uint) (*Client, error) {
|
|
client := &Client{
|
|
TenantID: tenantID,
|
|
Kind: "visitor",
|
|
SessionID: &sessionID,
|
|
Send: make(chan []byte, 256),
|
|
}
|
|
log.Printf("访客 WebSocket 连接: session=%d tenant=%d", sessionID, tenantID)
|
|
return upgrade(w, r, client)
|
|
}
|