chore(init): 初始化解忧商行项目并完成 Phase 0 基座
This commit is contained in:
yml2213
@@ -0,0 +1,150 @@
|
||||
package chathub
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"hfb_sys/backend/internal/middleware"
|
||||
"hfb_sys/backend/pkg/response"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
const heartbeatInterval = 30 * time.Second
|
||||
|
||||
type Handler struct {
|
||||
hub *Hub
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
func NewHandler(hub *Hub, logger *zap.Logger) *Handler {
|
||||
if logger == nil {
|
||||
logger = zap.NewNop()
|
||||
}
|
||||
return &Handler{hub: hub, logger: logger}
|
||||
}
|
||||
|
||||
// UserEvents 处理用户端 SSE 连接: GET /api/chats/events
|
||||
func (h *Handler) UserEvents(c *gin.Context) {
|
||||
value, ok := c.Get(middleware.ContextUserID)
|
||||
if !ok {
|
||||
response.Unauthorized(c, "缺少用户上下文")
|
||||
return
|
||||
}
|
||||
userID, ok := value.(uint64)
|
||||
if !ok {
|
||||
response.Unauthorized(c, "用户上下文无效")
|
||||
return
|
||||
}
|
||||
h.serveSSE(c, "user", userID)
|
||||
}
|
||||
|
||||
// AdminEvents 处理管理端 SSE 连接: GET /api/admin/chats/events
|
||||
func (h *Handler) AdminEvents(c *gin.Context) {
|
||||
value, ok := c.Get(middleware.ContextAdminID)
|
||||
if !ok {
|
||||
response.Unauthorized(c, "缺少管理员上下文")
|
||||
return
|
||||
}
|
||||
adminID, ok := value.(uint64)
|
||||
if !ok {
|
||||
response.Unauthorized(c, "管理员上下文无效")
|
||||
return
|
||||
}
|
||||
h.serveSSE(c, "admin", adminID)
|
||||
}
|
||||
|
||||
func (h *Handler) serveSSE(c *gin.Context, pType string, pID uint64) {
|
||||
startedAt := time.Now()
|
||||
disconnectReason := "handler_completed"
|
||||
var disconnectErr error
|
||||
ch := h.hub.Subscribe(pType, pID)
|
||||
h.logger.Info("SSE 连接建立", h.connectionLogFields(c, pType, pID)...)
|
||||
defer func() {
|
||||
h.hub.Unsubscribe(pType, pID, ch)
|
||||
fields := h.connectionLogFields(c, pType, pID)
|
||||
fields = append(fields,
|
||||
zap.String("disconnect_reason", disconnectReason),
|
||||
zap.Float64("lifetime_ms", float64(time.Since(startedAt).Microseconds())/1000),
|
||||
)
|
||||
if disconnectErr != nil {
|
||||
fields = append(fields, zap.Error(disconnectErr))
|
||||
}
|
||||
h.logger.Info("SSE 连接断开", fields...)
|
||||
}()
|
||||
|
||||
c.Header("Content-Type", "text/event-stream")
|
||||
c.Header("Cache-Control", "no-cache")
|
||||
c.Header("Connection", "keep-alive")
|
||||
c.Header("X-Accel-Buffering", "no")
|
||||
|
||||
// 发送初始连接确认
|
||||
if _, err := fmt.Fprintf(c.Writer, "event: connected\ndata: {\"ok\":true}\n\n"); err != nil {
|
||||
disconnectReason = "initial_write_error"
|
||||
disconnectErr = err
|
||||
return
|
||||
}
|
||||
c.Writer.Flush()
|
||||
|
||||
heartbeat := time.NewTicker(heartbeatInterval)
|
||||
defer heartbeat.Stop()
|
||||
|
||||
clientGone := c.Request.Context().Done()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-clientGone:
|
||||
disconnectReason = contextDisconnectReason(c.Request.Context().Err())
|
||||
return
|
||||
case <-heartbeat.C:
|
||||
if _, err := fmt.Fprintf(c.Writer, ":heartbeat\n\n"); err != nil {
|
||||
disconnectReason = "heartbeat_write_error"
|
||||
disconnectErr = err
|
||||
return
|
||||
}
|
||||
c.Writer.Flush()
|
||||
case event, ok := <-ch:
|
||||
if !ok {
|
||||
disconnectReason = "subscription_closed"
|
||||
return
|
||||
}
|
||||
data := MarshalEvent(event)
|
||||
if _, err := fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event.Type, data); err != nil {
|
||||
disconnectReason = "event_write_error"
|
||||
disconnectErr = err
|
||||
return
|
||||
}
|
||||
c.Writer.Flush()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Handler) connectionLogFields(c *gin.Context, pType string, pID uint64) []zap.Field {
|
||||
fields := []zap.Field{
|
||||
zap.String("request_id", middleware.GetRequestID(c)),
|
||||
zap.String("principal_type", pType),
|
||||
zap.Uint64("principal_id", pID),
|
||||
zap.String("client_ip", c.ClientIP()),
|
||||
zap.Int("active_sse_connections", h.hub.OnlineCount()),
|
||||
zap.Int("active_principal_type_connections", h.hub.OnlineCountByType(pType)),
|
||||
}
|
||||
if pType == "admin" {
|
||||
fields = append(fields, zap.Uint64("admin_id", pID))
|
||||
} else {
|
||||
fields = append(fields, zap.Uint64("user_id", pID))
|
||||
}
|
||||
return fields
|
||||
}
|
||||
|
||||
func contextDisconnectReason(err error) string {
|
||||
switch err {
|
||||
case context.Canceled:
|
||||
return "context_canceled"
|
||||
case context.DeadlineExceeded:
|
||||
return "context_deadline_exceeded"
|
||||
default:
|
||||
return "context_closed"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package chathub
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"hfb_sys/backend/internal/middleware"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
)
|
||||
|
||||
func TestSSELifecycleLogsConnectionCountReasonAndLifetime(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
core, observed := observer.New(zap.InfoLevel)
|
||||
hub := NewHub(nil)
|
||||
handler := NewHandler(hub, zap.New(core))
|
||||
|
||||
requestContext, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/admin/chats/events", nil).WithContext(requestContext)
|
||||
request.RemoteAddr = "42.49.131.69:34567"
|
||||
recorder := httptest.NewRecorder()
|
||||
ginContext, _ := gin.CreateTestContext(recorder)
|
||||
ginContext.Request = request
|
||||
ginContext.Set(middleware.ContextAdminID, uint64(88))
|
||||
ginContext.Set(middleware.ContextRequestID, "sse-test-request")
|
||||
|
||||
handler.AdminEvents(ginContext)
|
||||
|
||||
connected := observed.FilterMessage("SSE 连接建立").All()
|
||||
if len(connected) != 1 {
|
||||
t.Fatalf("connection logs = %d, want 1", len(connected))
|
||||
}
|
||||
connectedFields := connected[0].ContextMap()
|
||||
if connectedFields["admin_id"] != uint64(88) || connectedFields["active_sse_connections"] != int64(1) {
|
||||
t.Fatalf("unexpected connection fields: %v", connectedFields)
|
||||
}
|
||||
|
||||
disconnected := observed.FilterMessage("SSE 连接断开").All()
|
||||
if len(disconnected) != 1 {
|
||||
t.Fatalf("disconnection logs = %d, want 1", len(disconnected))
|
||||
}
|
||||
disconnectedFields := disconnected[0].ContextMap()
|
||||
if disconnectedFields["disconnect_reason"] != "context_canceled" {
|
||||
t.Fatalf("disconnect reason = %v, want context_canceled", disconnectedFields["disconnect_reason"])
|
||||
}
|
||||
if disconnectedFields["active_sse_connections"] != int64(0) {
|
||||
t.Fatalf("active connections after disconnect = %v, want 0", disconnectedFields["active_sse_connections"])
|
||||
}
|
||||
if _, ok := disconnectedFields["lifetime_ms"]; !ok {
|
||||
t.Fatalf("missing lifetime_ms field: %v", disconnectedFields)
|
||||
}
|
||||
if got := hub.OnlineCount(); got != 0 {
|
||||
t.Fatalf("online connection count = %d, want 0", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,206 @@
|
||||
package chathub
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sync"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// ChatEvent 是通过 SSE 推送给客户端的事件。
|
||||
type ChatEvent struct {
|
||||
Type string `json:"type"` // "new_message" | "conversation_updated" | "conversation_read"
|
||||
ConversationID uint64 `json:"conversation_id"`
|
||||
Message *MessageData `json:"message,omitempty"`
|
||||
// 以下字段仅在 conversation_read 事件中使用，标识谁在何时读取了会话。
|
||||
ReaderType string `json:"reader_type,omitempty"`
|
||||
ReaderID uint64 `json:"reader_id,omitempty"`
|
||||
ReadAt string `json:"read_at,omitempty"`
|
||||
}
|
||||
|
||||
// MessageData 是事件中携带的消息数据，与 chat.MessageDTO 对齐。
|
||||
type MessageData struct {
|
||||
ID uint64 `json:"id"`
|
||||
ConversationID uint64 `json:"conversation_id"`
|
||||
SenderType string `json:"sender_type"`
|
||||
SenderID uint64 `json:"sender_id"`
|
||||
SenderRole string `json:"sender_role"`
|
||||
SenderName string `json:"sender_name"`
|
||||
ContentType string `json:"content_type"`
|
||||
Content string `json:"content"`
|
||||
AttachmentURLS []string `json:"attachment_urls"`
|
||||
AdminAttentionType string `json:"admin_attention_type,omitempty"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
// principal 标识一个连接方。
|
||||
type principal struct {
|
||||
Type string // "user" or "admin"
|
||||
ID uint64
|
||||
}
|
||||
|
||||
// Hub 管理所有 SSE 连接。
|
||||
type Hub struct {
|
||||
mu sync.RWMutex
|
||||
clients map[principal]map[chan *ChatEvent]struct{}
|
||||
db *gorm.DB
|
||||
}
|
||||
|
||||
// NewHub 创建 Hub 实例。
|
||||
func NewHub(db *gorm.DB) *Hub {
|
||||
return &Hub{
|
||||
clients: make(map[principal]map[chan *ChatEvent]struct{}),
|
||||
db: db,
|
||||
}
|
||||
}
|
||||
|
||||
// Subscribe 注册一个 SSE 连接，返回事件 channel。
|
||||
func (h *Hub) Subscribe(pType string, pID uint64) <-chan *ChatEvent {
|
||||
ch := make(chan *ChatEvent, 16)
|
||||
key := principal{Type: pType, ID: pID}
|
||||
h.mu.Lock()
|
||||
if h.clients[key] == nil {
|
||||
h.clients[key] = make(map[chan *ChatEvent]struct{})
|
||||
}
|
||||
h.clients[key][ch] = struct{}{}
|
||||
h.mu.Unlock()
|
||||
return ch
|
||||
}
|
||||
|
||||
// Unsubscribe 注销 SSE 连接。
|
||||
func (h *Hub) Unsubscribe(pType string, pID uint64, ch <-chan *ChatEvent) {
|
||||
key := principal{Type: pType, ID: pID}
|
||||
h.mu.Lock()
|
||||
if clients, ok := h.clients[key]; ok {
|
||||
// 找到对应的发送 channel 并删除
|
||||
for sendCh := range clients {
|
||||
if sendCh == ch {
|
||||
delete(clients, sendCh)
|
||||
close(sendCh)
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(clients) == 0 {
|
||||
delete(h.clients, key)
|
||||
}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
}
|
||||
|
||||
// DisconnectUser 关闭指定用户当前进程内的全部实时连接。
|
||||
func (h *Hub) DisconnectUser(userID uint64) {
|
||||
h.disconnect("user", userID)
|
||||
}
|
||||
|
||||
func (h *Hub) disconnect(pType string, pID uint64) {
|
||||
key := principal{Type: pType, ID: pID}
|
||||
h.mu.Lock()
|
||||
clients := h.clients[key]
|
||||
delete(h.clients, key)
|
||||
for ch := range clients {
|
||||
close(ch)
|
||||
}
|
||||
h.mu.Unlock()
|
||||
}
|
||||
|
||||
// NotifyConversation 查询会话参与者，向所有在线参与者推送事件。
|
||||
func (h *Hub) NotifyConversation(conversationID uint64, event *ChatEvent) {
|
||||
if h.db == nil {
|
||||
return
|
||||
}
|
||||
// 查询会话参与者
|
||||
type participantRow struct {
|
||||
ParticipantType string
|
||||
ParticipantID uint64
|
||||
}
|
||||
var rows []participantRow
|
||||
h.db.Table("chat_participants").
|
||||
Select("participant_type, participant_id").
|
||||
Where("conversation_id = ?", conversationID).
|
||||
Find(&rows)
|
||||
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
seen := make(map[principal]struct{})
|
||||
for _, row := range rows {
|
||||
key := principal{Type: row.ParticipantType, ID: row.ParticipantID}
|
||||
if _, ok := seen[key]; ok {
|
||||
continue
|
||||
}
|
||||
seen[key] = struct{}{}
|
||||
if clients, ok := h.clients[key]; ok {
|
||||
for ch := range clients {
|
||||
select {
|
||||
case ch <- event:
|
||||
default:
|
||||
// channel 满了，丢弃事件避免阻塞
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// NotifyUser 直接向指定用户/Admin 推送事件。
|
||||
func (h *Hub) NotifyUser(pType string, pID uint64, event *ChatEvent) {
|
||||
key := principal{Type: pType, ID: pID}
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
if clients, ok := h.clients[key]; ok {
|
||||
for ch := range clients {
|
||||
select {
|
||||
case ch <- event:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// NotifyAllAdmins 向所有在线管理员（客服）推送事件。
|
||||
// 客服工作台的「未分配 / 全部」视图会展示当前客服并非参与者的会话，
|
||||
// 仅靠 NotifyConversation（只推参与者）无法实时刷新，故对所有在线客服广播。
|
||||
func (h *Hub) NotifyAllAdmins(event *ChatEvent) {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
for key, clients := range h.clients {
|
||||
if key.Type != "admin" {
|
||||
continue
|
||||
}
|
||||
for ch := range clients {
|
||||
select {
|
||||
case ch <- event:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MarshalEvent 将事件序列化为 SSE data 行。
|
||||
func MarshalEvent(event *ChatEvent) string {
|
||||
raw, _ := json.Marshal(event)
|
||||
return string(raw)
|
||||
}
|
||||
|
||||
// OnlineCount 返回当前在线连接数。
|
||||
func (h *Hub) OnlineCount() int {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
count := 0
|
||||
for _, clients := range h.clients {
|
||||
count += len(clients)
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
// OnlineCountByType 返回当前进程内指定主体类型的 SSE 连接数。
|
||||
func (h *Hub) OnlineCountByType(pType string) int {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
count := 0
|
||||
for key, clients := range h.clients {
|
||||
if key.Type == pType {
|
||||
count += len(clients)
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package chathub
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestDisconnectUserClosesAllConnections(t *testing.T) {
|
||||
hub := NewHub(nil)
|
||||
first := hub.Subscribe("user", 7)
|
||||
second := hub.Subscribe("user", 7)
|
||||
hub.Subscribe("user", 8)
|
||||
hub.Subscribe("admin", 9)
|
||||
|
||||
hub.DisconnectUser(7)
|
||||
for name, ch := range map[string]<-chan *ChatEvent{"first": first, "second": second} {
|
||||
select {
|
||||
case _, ok := <-ch:
|
||||
if ok {
|
||||
t.Fatalf("%s connection was not closed", name)
|
||||
}
|
||||
default:
|
||||
t.Fatalf("%s connection did not close immediately", name)
|
||||
}
|
||||
}
|
||||
if got := hub.OnlineCount(); got != 2 {
|
||||
t.Fatalf("online connection count = %d, want 2", got)
|
||||
}
|
||||
if got := hub.OnlineCountByType("user"); got != 1 {
|
||||
t.Fatalf("online user connection count = %d, want 1", got)
|
||||
}
|
||||
if got := hub.OnlineCountByType("admin"); got != 1 {
|
||||
t.Fatalf("online admin connection count = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user