Go后端网关服务详细分析

graph TD
    subgraph "Go后端网关服务架构"
        HTTP[HTTP请求] --> LB[负载均衡中间件]
        WS[WebSocket升级] --> WSH[WebSocket处理器]
        
        LB --> CORS[CORS中间件]
        CORS --> AUTH[JWT认证中间件]
        AUTH --> RBAC[RBAC权限中间件]
        RBAC --> TENANT[多租户中间件]
        TENANT --> ROUTER[路由分发器]
        
        ROUTER --> CHAT[聊天处理器]
        ROUTER --> VOICE[语音处理器]
        ROUTER --> ADMIN[管理处理器]
        ROUTER --> DOC[文档处理器]
        
        WSH --> VSESSION[语音会话管理]
        WSH --> WPROXY[WebSocket代理]
        
        CHAT --> ALGO[算法服务代理]
        VOICE --> ALGO
        ADMIN --> DB[(数据库)]
        DOC --> STORAGE[(对象存储)]
        
        VSESSION --> REDIS[(Redis缓存)]
        WPROXY --> ALGO
    end
    
    style HTTP fill:#e3f2fd
    style WS fill:#f3e5f5
    style ROUTER fill:#e8f5e8
    style ALGO fill:#fff3e0

// ChatRequest 聊天请求结构体
// 定义客户端发送的聊天请求格式
type ChatRequest struct {
    // 用户输入的消息内容，必填字段
    Message string `json:"message" binding:"required" example:"你好，请介绍一下VoiceHelper"`
    
    // 会话ID，用于维护对话上下文，可选
    ConversationID string `json:"conversation_id" example:"conv_123456789"`
    
    // 流ID，用于幂等性控制，可选
    StreamID string `json:"stream_id" example:"stream_abc123"`
    
    // 请求ID，用于请求追踪和幂等性，可选
    RequestID string `json:"request_id" example:"req_xyz789"`
    
    // 使用的AI模型，默认为gpt-3.5-turbo
    Model string `json:"model" example:"gpt-3.5-turbo"`
    
    // 生成温度，控制回复的随机性，范围0-2
    Temperature *float32 `json:"temperature" example:"0.7"`
    
    // 最大生成token数量，默认为2048
    MaxTokens *int `json:"max_tokens" example:"2048"`
}

// V2ChatHandlerSimple 简化版聊天处理器
// 主要负责请求转发和SSE流处理
type V2ChatHandlerSimple struct {
    // 算法服务的基础URL，用于请求转发
    algoServiceURL string
    
    // HTTP客户端，配置了连接池和超时设置
    httpClient *http.Client
    
    // 请求计数器，用于监控和限流
    requestCounter int64
    
    // 活跃请求映射，用于请求管理和取消
    activeRequests map[string]context.CancelFunc
    
    // 请求锁，保护并发访问
    requestMutex sync.RWMutex
}

// StreamChat 流式聊天处理函数
// @Summary 流式聊天API
// @Description 发送消息到AI助手并获取流式回复，使用Server-Sent Events协议
// @Tags Chat
// @Accept application/json
// @Produce text/event-stream
// @Param request body ChatRequest true "聊天请求参数"
// @Success 200 {string} string "SSE流式响应"
// @Failure 400 {object} ErrorResponse "请求参数错误"
// @Failure 500 {object} ErrorResponse "服务器内部错误"
// @Router /api/v2/chat/stream [post]
func (h *V2ChatHandlerSimple) StreamChat(c *gin.Context) {
    // === 第一阶段：请求解析和验证 ===
    
    // 1.1 解析JSON请求体
    var req ChatRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        logger.Error("Failed to bind JSON request", 
            zap.Error(err),
            zap.String("method", c.Request.Method),
            zap.String("path", c.Request.URL.Path))
        
        c.JSON(http.StatusBadRequest, gin.H{
            "error": "Invalid JSON format",
            "details": err.Error(),
        })
        return
    }
    
    // 1.2 业务参数验证
    if err := h.validateChatRequest(&req); err != nil {
        logger.Warn("Chat request validation failed",
            zap.Error(err),
            zap.String("message_preview", truncateString(req.Message, 50)))
        
        c.JSON(http.StatusBadRequest, gin.H{
            "error": "Validation failed",
            "details": err.Error(),
        })
        return
    }
    
    // 1.3 生成请求追踪ID（如果未提供）
    if req.RequestID == "" {
        req.RequestID = generateRequestID()
    }
    
    // === 第二阶段：SSE响应设置 ===
    
    // 2.1 设置Server-Sent Events响应头
    c.Header("Content-Type", "text/event-stream")       // SSE内容类型
    c.Header("Cache-Control", "no-cache")               // 禁用缓存
    c.Header("Connection", "keep-alive")                // 保持连接
    c.Header("Access-Control-Allow-Origin", "*")        // CORS设置
    c.Header("Access-Control-Allow-Headers", "Cache-Control")
    c.Header("X-Accel-Buffering", "no")                // 禁用Nginx缓冲
    
    // 2.2 创建可取消的上下文
    ctx, cancel := context.WithCancel(c.Request.Context())
    defer cancel()
    
    // 2.3 注册活跃请求，便于管理和取消
    h.requestMutex.Lock()
    if h.activeRequests == nil {
        h.activeRequests = make(map[string]context.CancelFunc)
    }
    h.activeRequests[req.RequestID] = cancel
    h.requestMutex.Unlock()
    
    // 请求结束时清理
    defer func() {
        h.requestMutex.Lock()
        delete(h.activeRequests, req.RequestID)
        h.requestMutex.Unlock()
    }()
    
    // === 第三阶段：请求转发到算法服务 ===
    
    // 3.1 构建算法服务请求体
    algoRequest := AlgoServiceRequest{
        Messages: []Message{
            {
                Role:    "user",
                Content: req.Message,
            },
        },
        ConversationID: req.ConversationID,
        StreamID:      req.StreamID,
        RequestID:     req.RequestID,
        Model:         getDefaultIfEmpty(req.Model, "gpt-3.5-turbo"),
        Temperature:   getDefaultIfNil(req.Temperature, 0.7),
        MaxTokens:    getDefaultIfNil(req.MaxTokens, 2048),
    }
    
    // 3.2 序列化请求体
    reqBody, err := json.Marshal(algoRequest)
    if err != nil {
        logger.Error("Failed to marshal algo request", 
            zap.Error(err),
            zap.String("request_id", req.RequestID))
        
        h.sendSSEError(c, "MARSHAL_ERROR", "Failed to process request")
        return
    }
    
    // 3.3 创建HTTP请求
    httpReq, err := http.NewRequestWithContext(
        ctx,
        "POST",
        h.algoServiceURL+"/query",
        bytes.NewBuffer(reqBody))
    if err != nil {
        logger.Error("Failed to create HTTP request", 
            zap.Error(err),
            zap.String("algo_url", h.algoServiceURL))
        
        h.sendSSEError(c, "REQUEST_CREATE_ERROR", "Failed to create request")
        return
    }
    
    // 3.4 设置请求头
    httpReq.Header.Set("Content-Type", "application/json")
    httpReq.Header.Set("Accept", "application/x-ndjson")  // NDJSON格式响应
    httpReq.Header.Set("X-Request-ID", req.RequestID)     // 请求追踪
    httpReq.Header.Set("User-Agent", "VoiceHelper-Gateway/2.0.0")
    
    // === 第四阶段：发送请求并处理响应 ===
    
    // 4.1 发送HTTP请求
    resp, err := h.httpClient.Do(httpReq)
    if err != nil {
        logger.Error("Failed to send request to algo service",
            zap.Error(err),
            zap.String("request_id", req.RequestID))
        
        h.sendSSEError(c, "ALGO_SERVICE_ERROR", "Algorithm service unavailable")
        return
    }
    defer resp.Body.Close()
    
    // 4.2 检查响应状态码
    if resp.StatusCode != http.StatusOK {
        logger.Error("Algo service returned non-200 status",
            zap.Int("status_code", resp.StatusCode),
            zap.String("request_id", req.RequestID))
        
        h.sendSSEError(c, "ALGO_SERVICE_ERROR", 
            fmt.Sprintf("Service returned status %d", resp.StatusCode))
        return
    }
    
    // 4.3 流式读取和转发响应
    if err := h.streamResponse(ctx, c, resp.Body, req.RequestID); err != nil {
        logger.Error("Failed to stream response",
            zap.Error(err),
            zap.String("request_id", req.RequestID))
        // 错误已在streamResponse中处理
    }
}

// streamResponse 流式处理算法服务响应
// @param ctx 请求上下文，用于取消控制
// @param c Gin上下文，用于SSE输出
// @param responseBody 算法服务的响应体
// @param requestID 请求ID，用于日志追踪
// @return error 处理过程中的错误
func (h *V2ChatHandlerSimple) streamResponse(
    ctx context.Context, 
    c *gin.Context, 
    responseBody io.ReadCloser, 
    requestID string,
) error {
    scanner := bufio.NewScanner(responseBody)
    scanner.Buffer(make([]byte, 64*1024), 1024*1024) // 64KB初始缓冲，1MB最大缓冲
    
    lineCount := 0
    startTime := time.Now()
    
    // 发送流开始事件
    h.sendSSEEvent(c, "stream_start", map[string]interface{}{
        "request_id": requestID,
        "timestamp": time.Now().Unix(),
    })
    
    for scanner.Scan() {
        select {
        case <-ctx.Done():
            // 请求被取消
            logger.Info("Stream cancelled by client",
                zap.String("request_id", requestID),
                zap.Int("lines_processed", lineCount))
            
            h.sendSSEEvent(c, "stream_cancelled", map[string]interface{}{
                "request_id": requestID,
                "reason": "Client disconnected",
            })
            return ctx.Err()
            
        default:
            // 处理响应行
            line := scanner.Text()
            if line == "" {
                continue // 跳过空行
            }
            
            lineCount++
            
            // 解析NDJSON行
            var response map[string]interface{}
            if err := json.Unmarshal([]byte(line), &response); err != nil {
                logger.Warn("Failed to parse response line",
                    zap.Error(err),
                    zap.String("line_preview", truncateString(line, 100)),
                    zap.String("request_id", requestID))
                continue // 跳过无效行
            }
            
            // 添加元数据
            response["request_id"] = requestID
            response["line_number"] = lineCount
            response["timestamp"] = time.Now().Unix()
            
            // 确定事件类型
            eventType := h.determineEventType(response)
            
            // 发送SSE事件
            if err := h.sendSSEEvent(c, eventType, response); err != nil {
                return fmt.Errorf("failed to send SSE event: %w", err)
            }
            
            // 强制刷新缓冲区，确保实时性
            if flusher, ok := c.Writer.(http.Flusher); ok {
                flusher.Flush()
            }
            
            // 流量控制，避免过快发送
            if lineCount%10 == 0 {
                time.Sleep(time.Millisecond) // 1ms间隔
            }
        }
    }
    
    // 检查扫描错误
    if err := scanner.Err(); err != nil {
        logger.Error("Scanner error during stream processing",
            zap.Error(err),
            zap.String("request_id", requestID))
        
        h.sendSSEError(c, "STREAM_ERROR", "Stream processing failed")
        return fmt.Errorf("scanner error: %w", err)
    }
    
    // 发送流结束事件
    duration := time.Since(startTime)
    h.sendSSEEvent(c, "stream_end", map[string]interface{}{
        "request_id":      requestID,
        "lines_processed": lineCount,
        "duration_ms":     duration.Milliseconds(),
        "timestamp":       time.Now().Unix(),
    })
    
    logger.Info("Stream completed successfully",
        zap.String("request_id", requestID),
        zap.Int("lines_processed", lineCount),
        zap.Duration("duration", duration))
    
    return nil
}

// sendSSEEvent 发送SSE事件
// @param c Gin上下文
// @param eventType 事件类型
// @param data 事件数据
// @return error 发送错误
func (h *V2ChatHandlerSimple) sendSSEEvent(c *gin.Context, eventType string, data interface{}) error {
    // 构建SSE格式消息
    eventData, err := json.Marshal(data)
    if err != nil {
        return fmt.Errorf("failed to marshal event data: %w", err)
    }
    
    // 写入SSE事件
    // 格式: event: eventType\ndata: jsonData\n\n
    if _, err := fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", eventType, string(eventData)); err != nil {
        return fmt.Errorf("failed to write SSE event: %w", err)
    }
    
    return nil
}

sequenceDiagram
    participant Client as 客户端
    participant Gateway as Go网关
    participant Algo as 算法服务
    participant LLM as AI模型
    
    Note over Client,LLM: 流式聊天完整调用链路
    
    Client->>Gateway: POST /api/v2/chat/stream
    Note right of Client: ChatRequest JSON
    
    Gateway->>Gateway: 1. JSON解析验证
    Gateway->>Gateway: 2. 设置SSE响应头
    Gateway->>Gateway: 3. 创建请求上下文
    
    Gateway->>Algo: POST /query (NDJSON)
    Note right of Gateway: 转发请求到算法服务
    
    Algo->>Algo: 4. RAG检索处理
    Algo-->>Gateway: retrieval_start event
    Gateway-->>Client: SSE: retrieval_start
    
    Algo->>Algo: 5. 文档向量检索
    Algo-->>Gateway: retrieval_progress event
    Gateway-->>Client: SSE: retrieval_progress
    
    Algo->>Algo: 6. 图谱推理检索
    Algo-->>Gateway: retrieval_result event
    Gateway-->>Client: SSE: retrieval_result
    
    Algo->>LLM: 7. 构建提示调用LLM
    Note right of Algo: 包含检索到的上下文
    
    LLM-->>Algo: generation_start
    Algo-->>Gateway: generation_start event
    Gateway-->>Client: SSE: generation_start
    
    loop 流式生成
        LLM-->>Algo: token chunk
        Algo-->>Gateway: generation_chunk event
        Gateway-->>Client: SSE: generation_chunk
        Note right of Client: 实时显示生成内容
    end
    
    LLM-->>Algo: generation complete
    Algo-->>Gateway: generation_done event
    Gateway-->>Client: SSE: generation_done
    
    Gateway-->>Client: SSE: stream_end
    
    Note over Client,LLM: 完整对话流程结束

// VoiceSessionConfig 语音会话配置
type VoiceSessionConfig struct {
    // 采样率，默认16000Hz
    SampleRate int `json:"sample_rate" example:"16000"`
    
    // 声道数，默认单声道
    Channels int `json:"channels" example:"1"`
    
    // 音频格式：pcm, opus, mp3
    Format string `json:"format" example:"pcm"`
    
    // 语言代码，默认zh-CN
    Language string `json:"language" example:"zh-CN"`
    
    // 是否启用VAD（语音活动检测）
    VADEnabled bool `json:"vad_enabled" example:"true"`
    
    // 是否启用噪声抑制
    NoiseSuppressionEnabled bool `json:"noise_suppression" example:"true"`
    
    // 是否启用回声消除
    EchoCancellationEnabled bool `json:"echo_cancellation" example:"false"`
}

// VoiceSessionMetrics 语音会话指标
type VoiceSessionMetrics struct {
    // 会话开始时间戳
    SessionStartTime time.Time `json:"session_start_time"`
    
    // 最后活动时间戳
    LastActivityTime time.Time `json:"last_activity_time"`
    
    // 总音频包数量
    AudioPackets int64 `json:"audio_packets"`
    
    // 总音频字节数
    AudioBytes int64 `json:"audio_bytes"`
    
    // 平均延迟（毫秒）
    AverageLatencyMS float64 `json:"average_latency_ms"`
    
    // ASR识别次数
    ASRRecognitions int `json:"asr_recognitions"`
    
    // TTS合成次数
    TTSSyntheses int `json:"tts_syntheses"`
    
    // 错误计数
    ErrorCount int `json:"error_count"`
}

// V2VoiceHandler 语音处理器V2版本
type V2VoiceHandler struct {
    // 算法服务URL
    algoServiceURL string
    
    // WebSocket升级器
    upgrader websocket.Upgrader
    
    // 活跃会话映射（session_id -> VoiceSession）
    activeConnections map[string]*VoiceSession
    
    // 读写互斥锁，保护并发访问
    mu sync.RWMutex
    
    // 会话清理定时器
    cleanupTicker *time.Ticker
    
    // 会话清理停止通道
    cleanupStop chan bool
    
    // 请求计数器（原子操作）
    requestCounter int64
    
    // 性能监控指标
    metrics *VoiceHandlerMetrics
}

// VoiceSession 语音会话结构
type VoiceSession struct {
    // 会话基本信息
    ID             string          `json:"session_id"`      // 唯一会话ID
    UserID         string          `json:"user_id"`         // 用户ID
    ConversationID string          `json:"conversation_id"` // 对话ID
    Status         string          `json:"status"`          // 会话状态: active, paused, ended
    CreatedAt      time.Time       `json:"created_at"`      // 创建时间
    LastActivity   time.Time       `json:"last_activity"`   // 最后活动时间
    
    // WebSocket连接
    Connection *websocket.Conn `json:"-"` // 客户端连接
    AlgoConn   *websocket.Conn `json:"-"` // 算法服务连接
    
    // 会话配置和指标
    Config  VoiceSessionConfig  `json:"config"`  // 音频配置
    Metrics VoiceSessionMetrics `json:"metrics"` // 性能指标
    
    // 控制和上下文
    CancelFunc context.CancelFunc `json:"-"` // 取消函数
    Context    context.Context    `json:"-"` // 会话上下文
    
    // 音频缓冲区
    audioBuffer      []byte        `json:"-"` // 音频数据缓冲
    transcriptBuffer string        `json:"-"` // 转录文本缓冲
    
    // 状态标志
    isProcessing bool          `json:"-"` // 是否正在处理
    lastError    error         `json:"-"` // 最后错误
    
    // 同步原语
    mu           sync.RWMutex  `json:"-"` // 会话级锁
}

// HandleWebSocket WebSocket语音处理入口函数
// @Summary WebSocket语音流处理
// @Description 建立WebSocket连接进行实时语音交互，支持全双工通信
// @Tags Voice
// @Accept application/json
// @Produce application/json
// @Param conversation_id query string false "会话ID" example:"conv_123"
// @Param language query string false "语言代码" example:"zh-CN"
// @Success 101 {object} VoiceSession "WebSocket连接升级成功"
// @Failure 400 {object} ErrorResponse "请求参数错误"
// @Failure 500 {object} ErrorResponse "服务器内部错误"
// @Router /api/v2/voice/stream [get]
func (h *V2VoiceHandler) HandleWebSocket(c *gin.Context) {
    // === 第一阶段：WebSocket连接升级 ===
    
    // 1.1 配置WebSocket升级器
    h.upgrader = websocket.Upgrader{
        ReadBufferSize:   4096,      // 4KB读缓冲
        WriteBufferSize:  4096,      // 4KB写缓冲
        HandshakeTimeout: 10 * time.Second, // 握手超时
        CheckOrigin: func(r *http.Request) bool {
            // 生产环境应该检查Origin
            return true // 允许所有来源
        },
        Subprotocols: []string{"voice-protocol-v2"}, // 支持的子协议
    }
    
    // 1.2 升级HTTP连接为WebSocket
    conn, err := h.upgrader.Upgrade(c.Writer, c.Request, nil)
    if err != nil {
        logger.Error("WebSocket upgrade failed",
            zap.Error(err),
            zap.String("remote_addr", c.Request.RemoteAddr),
            zap.String("user_agent", c.Request.UserAgent()))
        
        c.JSON(http.StatusBadRequest, gin.H{
            "error": "WebSocket upgrade failed",
            "details": err.Error(),
        })
        return
    }
    defer conn.Close() // 确保连接关闭
    
    // === 第二阶段：会话初始化 ===
    
    // 2.1 提取请求参数
    conversationID := c.Query("conversation_id")
    language := getDefaultIfEmpty(c.Query("language"), "zh-CN")
    userID := extractUserIDFromContext(c) // 从JWT或header提取
    
    // 2.2 生成唯一会话ID
    sessionID := h.generateSessionID()
    
    // 2.3 创建会话上下文
    ctx, cancel := context.WithCancel(context.Background())
    
    // 2.4 创建语音会话对象
    session := &VoiceSession{
        ID:             sessionID,
        UserID:         userID,
        ConversationID: conversationID,
        Status:         "initializing",
        CreatedAt:      time.Now(),
        LastActivity:   time.Now(),
        Connection:     conn,
        Context:        ctx,
        CancelFunc:     cancel,
        Config: VoiceSessionConfig{
            SampleRate:                 16000,
            Channels:                   1,
            Format:                     "pcm",
            Language:                   language,
            VADEnabled:                 true,
            NoiseSuppressionEnabled:    true,
            EchoCancellationEnabled:    false,
        },
        Metrics: VoiceSessionMetrics{
            SessionStartTime: time.Now(),
            LastActivityTime: time.Now(),
        },
    }
    
    // === 第三阶段：连接算法服务 ===
    
    // 3.1 建立与算法服务的WebSocket连接
    algoConn, err := h.connectToAlgoService(ctx, sessionID)
    if err != nil {
        logger.Error("Failed to connect to algo service",
            zap.Error(err),
            zap.String("session_id", sessionID),
            zap.String("algo_url", h.algoServiceURL))
        
        // 发送错误消息给客户端
        conn.WriteJSON(map[string]interface{}{
            "type":    "error",
            "error":   "Algorithm service unavailable",
            "code":    "ALGO_SERVICE_ERROR",
            "session_id": sessionID,
        })
        return
    }
    session.AlgoConn = algoConn
    defer algoConn.Close()
    
    // === 第四阶段：会话注册和管理 ===
    
    // 4.1 注册活跃会话
    h.mu.Lock()
    if h.activeConnections == nil {
        h.activeConnections = make(map[string]*VoiceSession)
    }
    h.activeConnections[sessionID] = session
    sessionCount := len(h.activeConnections)
    h.mu.Unlock()
    
    // 4.2 更新会话状态为活跃
    session.Status = "active"
    
    // 4.3 记录会话建立日志
    logger.Info("Voice session established",
        zap.String("session_id", sessionID),
        zap.String("user_id", userID),
        zap.String("conversation_id", conversationID),
        zap.Int("total_sessions", sessionCount),
        zap.String("client_ip", c.ClientIP()))
    
    // === 第五阶段：发送会话初始化确认 ===
    
    // 5.1 构建初始化消息
    initMessage := map[string]interface{}{
        "type":       "session_initialized",
        "session_id": sessionID,
        "config":     session.Config,
        "server_time": time.Now().Unix(),
        "capabilities": map[string]bool{
            "real_time_asr":    true,
            "voice_activity":   true,
            "noise_suppression": true,
            "stream_tts":       true,
            "cancel_request":   true,
        },
    }
    
    // 5.2 发送初始化消息
    if err := conn.WriteJSON(initMessage); err != nil {
        logger.Error("Failed to send initialization message",
            zap.Error(err),
            zap.String("session_id", sessionID))
        return
    }
    
    // === 第六阶段：启动消息处理协程 ===
    
    // 6.1 启动算法服务消息处理协程
    go h.handleAlgoServiceMessages(session)
    
    // 6.2 启动会话健康检查协程
    go h.monitorSessionHealth(session)
    
    // 6.3 处理客户端消息（主协程阻塞）
    h.handleClientMessages(session)
    
    // === 第七阶段：会话清理 ===
    h.cleanupSession(sessionID)
}

// handleClientMessages 处理来自客户端的WebSocket消息
// @param session 语音会话对象
func (h *V2VoiceHandler) handleClientMessages(session *VoiceSession) {
    defer func() {
        // 确保上下文取消和会话清理
        session.CancelFunc()
        session.Status = "ended"
        session.LastActivity = time.Now()
    }()
    
    // 设置连接参数
    session.Connection.SetReadLimit(1024 * 1024)           // 1MB读取限制
    session.Connection.SetReadDeadline(time.Now().Add(60 * time.Second)) // 60秒读超时
    session.Connection.SetPongHandler(func(appData string) error {
        // 处理pong消息，更新活动时间
        session.LastActivity = time.Now()
        session.Connection.SetReadDeadline(time.Now().Add(60 * time.Second))
        return nil
    })
    
    // 启动ping定时器
    pingTicker := time.NewTicker(30 * time.Second)
    defer pingTicker.Stop()
    
    go func() {
        // Ping协程，保持连接活跃
        for {
            select {
            case <-pingTicker.C:
                if err := session.Connection.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
                    return // 连接已断开
                }
            case <-session.Context.Done():
                return
            }
        }
    }()
    
    // 主消息处理循环
    for {
        select {
        case <-session.Context.Done():
            logger.Info("Session context cancelled",
                zap.String("session_id", session.ID))
            return
            
        default:
            // 读取WebSocket消息
            messageType, message, err := session.Connection.ReadMessage()
            if err != nil {
                if websocket.IsUnexpectedCloseError(err, 
                    websocket.CloseGoingAway, 
                    websocket.CloseAbnormalClosure,
                    websocket.CloseNormalClosure) {
                    
                    logger.Error("WebSocket unexpected close",
                        zap.Error(err),
                        zap.String("session_id", session.ID))
                } else {
                    logger.Info("WebSocket connection closed",
                        zap.String("session_id", session.ID),
                        zap.String("reason", err.Error()))
                }
                return
            }
            
            // 更新活动时间和指标
            session.LastActivity = time.Now()
            atomic.AddInt64(&session.Metrics.AudioPackets, 1)
            
            // 根据消息类型处理
            switch messageType {
            case websocket.TextMessage:
                // 处理JSON控制消息
                h.handleControlMessage(session, message)
                
            case websocket.BinaryMessage:
                // 处理音频数据
                h.handleAudioMessage(session, message)
                
            case websocket.CloseMessage:
                logger.Info("Received close message",
                    zap.String("session_id", session.ID))
                return
                
            default:
                logger.Warn("Unsupported message type",
                    zap.String("session_id", session.ID),
                    zap.Int("message_type", messageType))
            }
        }
    }
}

// handleAudioMessage 处理音频消息
// @param session 语音会话
// @param audioData 音频字节数据
func (h *V2VoiceHandler) handleAudioMessage(session *VoiceSession, audioData []byte) {
    // 验证音频数据
    if len(audioData) == 0 {
        return
    }
    
    // 更新指标
    atomic.AddInt64(&session.Metrics.AudioBytes, int64(len(audioData)))
    
    // 构建音频消息发送给算法服务
    audioMessage := map[string]interface{}{
        "type":            "audio_chunk",
        "session_id":      session.ID,
        "conversation_id": session.ConversationID,
        "audio_chunk":     base64.StdEncoding.EncodeToString(audioData),
        "timestamp":       time.Now().UnixNano() / 1e6, // 毫秒时间戳
        "sequence":        session.Metrics.AudioPackets,
        "format":          session.Config.Format,
        "sample_rate":     session.Config.SampleRate,
        "channels":        session.Config.Channels,
    }
    
    // 发送到算法服务
    if err := session.AlgoConn.WriteJSON(audioMessage); err != nil {
        logger.Error("Failed to forward audio to algo service",
            zap.Error(err),
            zap.String("session_id", session.ID))
        
        // 发送错误给客户端
        session.Connection.WriteJSON(map[string]interface{}{
            "type":    "error",
            "error":   "Failed to process audio",
            "code":    "AUDIO_FORWARD_ERROR",
            "session_id": session.ID,
        })
        return
    }
    
    // 记录音频处理日志（每100个包记录一次）
    if session.Metrics.AudioPackets%100 == 0 {
        logger.Debug("Audio processing progress",
            zap.String("session_id", session.ID),
            zap.Int64("packets", session.Metrics.AudioPackets),
            zap.Int64("bytes", session.Metrics.AudioBytes))
    }
}

sequenceDiagram
    participant C as 客户端
    participant GW as Go网关
    participant AS as 算法服务
    participant ASR as ASR服务
    participant LLM as LLM服务
    participant TTS as TTS服务
    
    Note over C,TTS: WebSocket语音交互完整流程
    
    C->>GW: WebSocket升级请求
    GW->>GW: 连接升级和会话创建
    GW->>AS: 建立WebSocket连接
    GW->>C: session_initialized
    
    Note over C,TTS: 音频流处理阶段
    
    loop 实时音频处理
        C->>GW: 二进制音频数据
        GW->>AS: 转发音频chunk
        
        AS->>ASR: 实时语音识别
        ASR-->>AS: 部分识别结果
        AS-->>GW: asr_partial事件
        GW-->>C: 转发识别结果
        
        Note right of C: 实时显示转录文本
    end
    
    Note over C,TTS: 完整句子处理阶段
    
    ASR-->>AS: 最终识别结果
    AS-->>GW: asr_final事件
    GW-->>C: 最终识别文本
    
    AS->>AS: RAG知识检索
    AS-->>GW: retrieval_progress
    GW-->>C: 检索进度更新
    
    AS->>LLM: 构建提示生成回复
    LLM-->>AS: 流式文本回复
    AS-->>GW: llm_response_chunk
    GW-->>C: AI回复文本
    
    AS->>TTS: 语音合成请求
    loop 流式TTS
        TTS-->>AS: 音频数据块
        AS-->>GW: tts_audio事件
        GW-->>C: 音频数据
        Note right of C: 播放AI语音回复
    end
    
    AS-->>GW: tts_complete
    GW-->>C: 语音合成完成
    
    Note over C,TTS: 会话结束清理
    
    C->>GW: 断开WebSocket
    GW->>GW: 会话清理
    GW->>AS: 关闭算法服务连接

// AuthMiddleware JWT认证中间件
// 负责验证JWT token、提取用户信息、处理token刷新
type AuthMiddleware struct {
    // JWT密钥，用于签名验证
    jwtSecret string
    
    // Token过期时间配置
    accessTokenTTL  time.Duration // 访问token有效期
    refreshTokenTTL time.Duration // 刷新token有效期
    
    // Redis客户端，用于token黑名单和会话管理
    redisClient *redis.Client
    
    // 白名单路径，不需要认证的路径
    whitelist map[string]bool
    
    // 用户服务客户端，用于获取用户信息
    userService UserServiceInterface
}

// Handle JWT认证中间件处理函数
// @Summary JWT token验证和用户信息提取
// @Description 验证Authorization header中的JWT token，提取用户信息并注入上下文
// @Security BearerAuth
func (m *AuthMiddleware) Handle() gin.HandlerFunc {
    return func(c *gin.Context) {
        // === 第一阶段：路径白名单检查 ===
        
        // 1.1 检查是否在白名单中
        path := c.Request.URL.Path
        if m.isWhitelisted(path) {
            logger.Debug("Path in whitelist, skipping auth",
                zap.String("path", path))
            c.Next()
            return
        }
        
        // === 第二阶段：Token提取和验证 ===
        
        // 2.1 从Header提取token
        tokenString := m.extractTokenFromHeader(c)
        if tokenString == "" {
            logger.Warn("Missing authorization token",
                zap.String("path", path),
                zap.String("method", c.Request.Method),
                zap.String("client_ip", c.ClientIP()))
            
            c.JSON(http.StatusUnauthorized, gin.H{
                "error": "Missing authorization token",
                "code":  "TOKEN_MISSING",
            })
            c.Abort()
            return
        }
        
        // 2.2 验证token格式和签名
        claims, err := m.validateToken(tokenString)
        if err != nil {
            logger.Warn("Invalid JWT token",
                zap.Error(err),
                zap.String("path", path),
                zap.String("token_preview", maskToken(tokenString)))
            
            c.JSON(http.StatusUnauthorized, gin.H{
                "error": "Invalid token",
                "code":  "TOKEN_INVALID",
                "details": err.Error(),
            })
            c.Abort()
            return
        }
        
        // === 第三阶段：Token黑名单检查 ===
        
        // 3.1 检查token是否在黑名单中（已注销）
        if isBlacklisted, err := m.isTokenBlacklisted(claims.JTI); err != nil {
            logger.Error("Failed to check token blacklist",
                zap.Error(err),
                zap.String("jti", claims.JTI))
            
            c.JSON(http.StatusInternalServerError, gin.H{
                "error": "Authentication service unavailable",
                "code":  "AUTH_SERVICE_ERROR",
            })
            c.Abort()
            return
        } else if isBlacklisted {
            logger.Warn("Token is blacklisted",
                zap.String("jti", claims.JTI),
                zap.String("user_id", claims.UserID))
            
            c.JSON(http.StatusUnauthorized, gin.H{
                "error": "Token has been revoked",
                "code":  "TOKEN_REVOKED",
            })
            c.Abort()
            return
        }
        
        // === 第四阶段：用户信息获取和验证 ===
        
        // 4.1 从缓存或数据库获取用户信息
        user, err := m.getUserInfo(claims.UserID)
        if err != nil {
            logger.Error("Failed to get user info",
                zap.Error(err),
                zap.String("user_id", claims.UserID))
            
            c.JSON(http.StatusUnauthorized, gin.H{
                "error": "User not found or inactive",
                "code":  "USER_NOT_FOUND",
            })
            c.Abort()
            return
        }
        
        // 4.2 检查用户状态
        if user.Status != "active" {
            logger.Warn("User account is not active",
                zap.String("user_id", claims.UserID),
                zap.String("status", user.Status))
            
            c.JSON(http.StatusUnauthorized, gin.H{
                "error": "User account is disabled",
                "code":  "USER_DISABLED",
            })
            c.Abort()
            return
        }
        
        // === 第五阶段：Token刷新检查 ===
        
        // 5.1 检查token是否即将过期（剩余时间 < 15分钟）
        refreshThreshold := time.Now().Add(15 * time.Minute)
        if claims.ExpiresAt.Time.Before(refreshThreshold) {
            // 设置响应头提示客户端刷新token
            c.Header("X-Token-Refresh-Required", "true")
            c.Header("X-Token-Expires-At", strconv.FormatInt(claims.ExpiresAt.Unix(), 10))
            
            logger.Info("Token refresh recommended",
                zap.String("user_id", claims.UserID),
                zap.Time("expires_at", claims.ExpiresAt.Time))
        }
        
        // === 第六阶段：上下文注入 ===
        
        // 6.1 将认证信息注入Gin上下文
        c.Set("user_id", user.UserID)
        c.Set("username", user.Username)
        c.Set("user_roles", user.Roles)
        c.Set("tenant_id", claims.TenantID)
        c.Set("session_id", claims.SessionID)
        c.Set("token_jti", claims.JTI)
        
        // 6.2 设置请求追踪信息
        c.Set("auth_method", "jwt")
        c.Set("auth_timestamp", time.Now())
        
        // 6.3 记录认证成功日志
        logger.Debug("Authentication successful",
            zap.String("user_id", user.UserID),
            zap.String("path", path),
            zap.String("method", c.Request.Method))
        
        // 继续处理请求
        c.Next()
    }
}

// validateToken 验证JWT token并返回claims
// @param tokenString JWT token字符串
// @return (*JWTClaims, error) JWT claims和错误
func (m *AuthMiddleware) validateToken(tokenString string) (*JWTClaims, error) {
    // 解析JWT token
    token, err := jwt.ParseWithClaims(tokenString, &JWTClaims{}, func(token *jwt.Token) (interface{}, error) {
        // 验证签名算法
        if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
            return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
        }
        return []byte(m.jwtSecret), nil
    })
    
    if err != nil {
        return nil, fmt.Errorf("failed to parse token: %w", err)
    }
    
    // 检查token是否有效
    if !token.Valid {
        return nil, fmt.Errorf("token is invalid")
    }
    
    // 提取claims
    claims, ok := token.Claims.(*JWTClaims)
    if !ok {
        return nil, fmt.Errorf("invalid token claims")
    }
    
    // 验证必要字段
    if claims.UserID == "" {
        return nil, fmt.Errorf("missing user_id in token")
    }
    
    if claims.JTI == "" {
        return nil, fmt.Errorf("missing jti in token")
    }
    
    // 验证过期时间
    if claims.ExpiresAt != nil && claims.ExpiresAt.Time.Before(time.Now()) {
        return nil, fmt.Errorf("token has expired")
    }
    
    return claims, nil
}

// JWTClaims JWT声明结构
type JWTClaims struct {
    UserID    string `json:"user_id"`    // 用户ID
    Username  string `json:"username"`   // 用户名
    TenantID  string `json:"tenant_id"`  // 租户ID
    SessionID string `json:"session_id"` // 会话ID
    Roles     []string `json:"roles"`    // 用户角色
    Scopes    []string `json:"scopes"`   // 权限范围
    
    jwt.RegisteredClaims // 标准JWT字段
}