完成客服工作台闭环

server/internal/ws/ws.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/gorilla/websocket"
+	"kefu-sys/server/internal/model"
 )
 
 var upgrader = websocket.Upgrader{
@@ -32,6 +33,11 @@ type Event struct {
 	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
@@ -118,17 +124,59 @@ func (h *Hub) BroadcastToTenantStaff(tenantID uint, message []byte) {
 	}
 }
 
+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)
+		}
+	}
+}
+
+func handleClientEvent(client *Client, event ClientEvent) {
+	if client.Kind != "agent" || 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.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, nil)
+	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 {
-		if _, _, err := client.Conn.ReadMessage(); err != nil {
+		_, message, err := client.Conn.ReadMessage()
+		if err != nil {
 			return
 		}
+		var event ClientEvent
+		if json.Unmarshal(message, &event) == nil {
+			handleClientEvent(client, event)
+		}
 	}
 }
 

server/internal/ws/ws_test.go

@@ -39,3 +39,31 @@ func TestBroadcastToSessionRestrictsRecipients(t *testing.T) {
 		}
 	}
 }
+
+func TestBroadcastToVisitorRestrictsSession(t *testing.T) {
+	hub := NewHub()
+	go hub.Run()
+
+	sessionID := uint(21)
+	otherSessionID := uint(22)
+	visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &sessionID, Send: make(chan []byte, 1)}
+	otherVisitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &otherSessionID, Send: make(chan []byte, 1)}
+	agent := &Client{TenantID: 1, Kind: "agent", UserID: 1, Role: "agent", Send: make(chan []byte, 1)}
+	for _, client := range []*Client{visitor, otherVisitor, agent} {
+		hub.register <- client
+	}
+
+	hub.BroadcastToVisitor(1, sessionID, []byte(`{"type":"typing"}`))
+	select {
+	case <-visitor.Send:
+	case <-time.After(time.Second):
+		t.Fatal("目标访客未收到输入状态")
+	}
+	for _, client := range []*Client{otherVisitor, agent} {
+		select {
+		case <-client.Send:
+			t.Fatalf("无关客户端收到访客状态：%+v", client)
+		default:
+		}
+	}
+}