nl-im-websocket-demo/controller/websocket_controller.go

/**
 * package controller
 * 作用：处理核心业务逻辑，包括 WebSocket 连接管理、消息转发、Redis 集群通信以及 API 接口实现。
 * 该包是连接网络层（Gin）与数据层（Gorm/Redis）的桥梁。
 */
package controller

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/go-redis/redis/v8"
	"github.com/golang-jwt/jwt/v5"
	"github.com/gorilla/websocket"
	"gorm.io/gorm"
	"xk-websocket/models"
	"xk-websocket/utils"
)

// JWT相关常量（与Laravel JWTService保持一致）
const (
	AdminJWTSecretKey = "xk_admin_secret_key" // Admin JWT密钥
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

/**
 * WebSocketController
 * 结构体定义：WebSocket 服务的主控制器。
 * 包含：
 * - 客户端连接池 (Clients)
 * - 读写锁 (ClientsMux) 保护连接池安全
 * - Redis 客户端 (RedisCli) 用于集群通信
 * - 数据库连接 (DB)
 * - Token验证相关 (AuthRedisCli, ClientInfoMap)
 */
type WebSocketController struct {
	Clients       map[string]*websocket.Conn
	ClientsMux    sync.RWMutex
	WriteMutex    sync.Mutex // 添加写锁
	RedisCli      *redis.Client
	AuthRedisCli  *redis.Client                 // 用于Token验证的Redis客户端 (DB 1)
	ClientInfos   map[string]*models.ClientInfo // 客户端认证信息映射
	ClientInfoMux sync.RWMutex                  // 客户端信息锁
	NodeID        string
	Port          string
	RedisCtx      context.Context
	Logger        *log.Logger
	DB            *gorm.DB // 数据库连接
}

/**
 * NewWebSocketController
 * 功能：创建并初始化 WebSocketController 实例。
 * @param db *gorm.DB 数据库连接实例
 */
func NewWebSocketController(db *gorm.DB) *WebSocketController {
	return &WebSocketController{
		Clients:     make(map[string]*websocket.Conn),
		ClientInfos: make(map[string]*models.ClientInfo),
		RedisCtx:    context.Background(),
		DB:          db, // 初始化数据库连接
	}
}

/**
 * ConfigureSystem
 * 功能：加载系统配置，如 NodeID 和端口，并设置日志格式。
 * 也会启动节点注册协程。
 */
func (c *WebSocketController) ConfigureSystem() {
	c.NodeID = utils.GetEnv("NODE_ID", "local")

	// 只有当端口未设置时才从环境变量获取
	if c.Port == "" {
		c.Port = utils.GetEnv("PORT", "12080")
		log.Printf("📡📡📡📡📡📡📡📡📡📡📡📡📡📡📡📡 使用环境变量设置端口: %s", c.Port)
	}

	log.SetPrefix(fmt.Sprintf("[Node:%s] ", c.NodeID))
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)
	c.configureLogger()

	// 必须先初始化 Redis 客户端
	c.InitRedisClient()

	// 注册节点到Redis
	go c.registerNode() // 现在 RedisCli 已初始化
}

/**
 * registerNode
 * 功能：定时将当前节点的 IP 和端口注册到 Redis 中，以便其他服务发现。
 * 这是一个心跳机制，每 20 秒执行一次。
 */
func (c *WebSocketController) registerNode() {
	// 添加空值检查
	if c.RedisCli == nil {
		log.Println("⚠️ Redis客户端未初始化，节点注册已取消")
		return
	}

	for {
		key := fmt.Sprintf("websocket:nodes:%s", c.NodeID)
		value := fmt.Sprintf("%s:%s", utils.GetOutboundIP(), c.Port)

		if err := c.RedisCli.Set(c.RedisCtx, key, value, 30*time.Second).Err(); err != nil {
			log.Printf("⚠️ 节点注册失败: %v", err)
		} else {
			log.Printf("📌📌📌📌📌📌📌📌📌📌📌📌📌📌📌📌 节点已注册: %s = %s", key, value)
		}

		time.Sleep(20 * time.Second)
	}
}

/**
 * configureLogger
 * 功能：初始化日志目录和写入器配置。
 */
func (c *WebSocketController) configureLogger() {
	logDir := utils.GetEnv("LOG_DIR", "./logs")
	if err := os.MkdirAll(logDir, 0755); err != nil {
		log.Fatalf("❌❌❌❌❌❌❌❌ 创建日志目录失败: %v", err)
	}

	logFreq := utils.GetEnv("LOG_ROTATE_FREQ", "daily")
	if c.Logger == nil {
		c.Logger = log.New(os.Stdout, "", log.LstdFlags|log.Lmicroseconds)
	}

	c.Logger.SetPrefix(fmt.Sprintf("[Node:%s] ", c.NodeID))
	log.SetOutput(c.getLogWriter(logDir, logFreq))
}

/**
 * getLogWriter
 * 功能：工厂方法，获取 DailyFileWriter 实例。
 */
func (c *WebSocketController) getLogWriter(logDir, freq string) *utils.DailyFileWriter {
	return utils.NewDailyFileWriter(logDir, freq, c.Logger, c.NodeID)
}

/**
 * InitRedisClient
 * 功能：建立 Redis 连接并验证连通性。
 * 初始化两个Redis客户端：
 * - RedisCli: DB 0，用于WebSocket集群通信
 * - AuthRedisCli: DB 1，用于Token验证（与Laravel保持一致）
 */
func (c *WebSocketController) InitRedisClient() {
	redisAddr := utils.GetEnv("REDIS_ADDR", "localhost:6379")
	redisPassword := utils.GetEnv("REDIS_PASSWORD", "")

	// 初始化主Redis客户端 (DB 0) 用于集群通信
	c.RedisCli = redis.NewClient(&redis.Options{
		Addr:     redisAddr,
		Password: redisPassword,
		DB:       0,
	})

	if err := c.checkRedisConnection(); err != nil {
		log.Fatalf("❌❌❌❌❌❌❌❌ Redis连接失败: %v", err)
	}

	// 初始化Token验证Redis客户端 (DB 1) 用于Token验证
	c.AuthRedisCli = redis.NewClient(&redis.Options{
		Addr:     redisAddr,
		Password: redisPassword,
		DB:       1, // 与Laravel cache连接保持一致
	})

	if err := c.checkAuthRedisConnection(); err != nil {
		log.Fatalf("❌❌❌❌❌❌❌❌ Token验证Redis连接失败: %v", err)
	}

	log.Println("✅ Token验证Redis客户端初始化成功 (DB 1)")
}

// 检查Redis连接
func (c *WebSocketController) checkRedisConnection() error {
	_, err := c.RedisCli.Ping(c.RedisCtx).Result()
	return err
}

// 检查Token验证Redis连接
func (c *WebSocketController) checkAuthRedisConnection() error {
	_, err := c.AuthRedisCli.Ping(c.RedisCtx).Result()
	return err
}

/**
 * ValidateToken
 * 功能：验证Token是否有效
 * 逻辑：从Redis DB 1中查询 xk_user_{token} 是否存在
 * @param token string 待验证的Token
 * @return (userInfo string, isValid bool) 用户信息JSON和验证结果
 */
func (c *WebSocketController) ValidateToken(token string) (string, bool) {
	if token == "" {
		log.Printf("⚠️ Token验证失败: Token为空")
		return "", false
	}

	// 构造Redis Key: xk_user_{token}
	redisKey := models.TokenKeyPrefix + token

	// 从Redis DB 1中查询
	userInfo, err := c.AuthRedisCli.Get(c.RedisCtx, redisKey).Result()
	if err != nil {
		if err == redis.Nil {
			log.Printf("⚠️ Token验证失败: Token不存在或已过期 | Key=%s", redisKey)
		} else {
			log.Printf("❌ Token验证错误: %v | Key=%s", err, redisKey)
		}
		return "", false
	}

	log.Printf("✅ Token验证成功 | Key=%s", redisKey)
	return userInfo, true
}

/**
 * AdminJWTClaims
 * 结构体：Admin JWT Token的Claims结构
 * 与Laravel JWTService的payload结构保持一致
 */
type AdminJWTClaims struct {
	Data struct {
		ID int `json:"id"` // 用户ID
	} `json:"data"`
	jwt.RegisteredClaims
}

/**
 * ValidateAdminToken
 * 功能：验证Admin后台的JWT Token是否有效
 * 逻辑：
 * 1. JWT解码获取 payload.data.id (user_id)
 * 2. 用 xk_login_{user_id} 在 Redis DB 0 查询
 * @param token string 待验证的JWT Token
 * @return (userInfo string, isValid bool) 用户信息JSON和验证结果
 */
func (c *WebSocketController) ValidateAdminToken(token string) (string, bool) {
	if token == "" {
		log.Printf("⚠️ Admin Token验证失败: Token为空")
		return "", false
	}

	// 解析JWT Token
	claims := &AdminJWTClaims{}
	parsedToken, err := jwt.ParseWithClaims(token, claims, func(t *jwt.Token) (interface{}, error) {
		// 验证签名算法
		if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
			return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"])
		}
		return []byte(AdminJWTSecretKey), nil
	})

	if err != nil {
		log.Printf("⚠️ Admin Token解析失败: %v | Token=%s", err, maskToken(token))
		return "", false
	}

	if !parsedToken.Valid {
		log.Printf("⚠️ Admin Token无效 | Token=%s", maskToken(token))
		return "", false
	}

	// 获取用户ID
	userID := claims.Data.ID
	if userID == 0 {
		log.Printf("⚠️ Admin Token验证失败: 无法获取用户ID | Token=%s", maskToken(token))
		return "", false
	}

	// 构造Redis Key: xk_login_{user_id}
	redisKey := fmt.Sprintf("%s%d", models.AdminTokenKeyPrefix, userID)

	// 从Redis DB 0中查询（使用主Redis客户端）
	userInfo, err := c.RedisCli.Get(c.RedisCtx, redisKey).Result()
	if err != nil {
		if err == redis.Nil {
			log.Printf("⚠️ Admin Token验证失败: 用户登录信息不存在或已过期 | Key=%s", redisKey)
		} else {
			log.Printf("❌ Admin Token验证错误: %v | Key=%s", err, redisKey)
		}
		return "", false
	}

	log.Printf("✅ Admin Token验证成功 | UserID=%d | Key=%s", userID, redisKey)
	return userInfo, true
}

/**
 * ValidateTokenByUserType
 * 功能：根据用户类型选择对应的Token验证方法
 * @param token string 待验证的Token
 * @param userType string 用户类型 (doctor/user)
 * @return (userInfo string, isValid bool) 用户信息JSON和验证结果
 */
func (c *WebSocketController) ValidateTokenByUserType(token string, userType string) (string, bool) {
	if userType == models.UserTypeDoctor {
		// 后台用户（医生端PC）：使用JWT Token验证
		log.Printf("🔐 使用Admin Token验证方式 | UserType=%s", userType)
		return c.ValidateAdminToken(token)
	}
	// 小程序用户：使用原始Token验证
	log.Printf("🔐 使用Mobile Token验证方式 | UserType=%s", userType)
	return c.ValidateToken(token)
}

/**
 * SaveClientInfo
 * 功能：保存客户端认证信息
 * @param clientID string 客户端ID
 * @param info *models.ClientInfo 客户端信息
 */
func (c *WebSocketController) SaveClientInfo(clientID string, info *models.ClientInfo) {
	c.ClientInfoMux.Lock()
	defer c.ClientInfoMux.Unlock()
	c.ClientInfos[clientID] = info
}

/**
 * GetClientInfo
 * 功能：获取客户端认证信息
 * @param clientID string 客户端ID
 * @return *models.ClientInfo 客户端信息
 */
func (c *WebSocketController) GetClientInfo(clientID string) *models.ClientInfo {
	c.ClientInfoMux.RLock()
	defer c.ClientInfoMux.RUnlock()
	return c.ClientInfos[clientID]
}

/**
 * RemoveClientInfo
 * 功能：移除客户端认证信息
 * @param clientID string 客户端ID
 */
func (c *WebSocketController) RemoveClientInfo(clientID string) {
	c.ClientInfoMux.Lock()
	defer c.ClientInfoMux.Unlock()
	delete(c.ClientInfos, clientID)
}

/**
 * SendAuthResponse
 * 功能：向客户端发送认证响应
 * @param clientID string 客户端ID
 * @param status string 认证状态 (success/failed/token_expired)
 * @param message string 消息说明
 */
func (c *WebSocketController) SendAuthResponse(clientID string, status string, message string) {
	c.ClientsMux.RLock()
	conn, exists := c.Clients[clientID]
	c.ClientsMux.RUnlock()

	if !exists {
		return
	}

	authResp := models.AuthResponse{
		AuthStatus: status,
		Message:    message,
		ClientID:   clientID,
	}

	c.WriteMutex.Lock()
	defer c.WriteMutex.Unlock()

	if err := conn.WriteJSON(authResp); err != nil {
		log.Printf("❌ 发送认证响应失败: %v | ClientID=%s", err, clientID)
	}
}

/**
 * maskToken
 * 功能：对Token进行脱敏处理，用于日志输出
 * @param token string 原始Token
 * @return string 脱敏后的Token
 */
func maskToken(token string) string {
	if len(token) <= 8 {
		return "***"
	}
	return token[:4] + "****" + token[len(token)-4:]
}

/**
 * startTokenCheckTimer
 * 功能：启动定期Token检查定时器
 * 每5分钟检查一次Token是否仍然有效
 * 根据用户类型使用不同的验证方法
 * @param clientID string 客户端ID
 */
func (c *WebSocketController) startTokenCheckTimer(clientID string) {
	ticker := time.NewTicker(5 * time.Minute)
	defer ticker.Stop()

	for range ticker.C {
		// 检查客户端是否仍然连接
		c.ClientsMux.RLock()
		_, exists := c.Clients[clientID]
		c.ClientsMux.RUnlock()

		if !exists {
			log.Printf("🔌 客户端已断开，停止Token检查: ClientID=%s", clientID)
			return
		}

		// 获取客户端信息
		clientInfo := c.GetClientInfo(clientID)
		if clientInfo == nil || clientInfo.Token == "" {
			log.Printf("⚠️ 客户端信息不存在，停止Token检查: ClientID=%s", clientID)
			return
		}

		// 根据用户类型选择验证方法
		_, isValid := c.ValidateTokenByUserType(clientInfo.Token, clientInfo.UserType)
		if !isValid {
			log.Printf("⚠️ Token已过期，通知客户端重新认证: ClientID=%s | UserType=%s", clientID, clientInfo.UserType)
			c.SendAuthResponse(clientID, "token_expired", "Token已过期，请重新登录")

			// 更新客户端认证状态
			clientInfo.IsAuth = false
			c.SaveClientInfo(clientID, clientInfo)
			return
		}

		// 更新最后检查时间
		clientInfo.LastCheckTime = time.Now()
		c.SaveClientInfo(clientID, clientInfo)
		log.Printf("✅ Token定期检查通过: ClientID=%s | UserType=%s", clientID, clientInfo.UserType)
	}
}

/**
 * PrintStartupInfo
 * 功能：在控制台打印详细的服务启动信息，包括节点ID、端口、Redis状态及支持的消息类型。
 */
func (c *WebSocketController) PrintStartupInfo() {
	hostname, _ := os.Hostname()
	log.Printf("🚀🚀🚀🚀🚀🚀🚀🚀 WebSocket服务启动: NodeID=%s", c.NodeID)
	log.Printf("🌐🌐🌐🌐🌐🌐🌐🌐 监听端口: %s", c.Port)
	log.Printf("📡📡📡📡📡📡📡📡 Redis地址: %s", utils.GetEnv("REDIS_ADDR", "localhost:6379"))
	log.Printf("💻💻💻💻💻💻💻💻 主机: %s", hostname)
	log.Printf("🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒 启动时间: %s", time.Now().Format("2006-01-02 15:04:05"))
	log.Printf("🔔🔔🔔🔔🔔🔔🔔🔔 支持消息类型: \n  0:文本\n  1:图片\n  2:音频\n  3:视频\n  4:处方\n  5:病例\n  6:视频通话\n  7:语音通话\n  8:文件消息")
	log.Printf("🔑🔑🔑🔑🔑🔑🔑🔑 用户绑定键: \n  %s\n  %s", models.ClientUserKey, models.UserClientKey)
	log.Printf("📬📬📬📬📬📬📬📬 新增接口: POST /send-to-user (直接通过用户ID发送消息)")
	log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 集群节点注册: %s:%s", utils.GetOutboundIP(), c.Port)
	log.Println("🔗🔗🔗🔗🔗🔗🔗🔗 等待客户端连接...")
}

/**
 * HealthHandler
 * 功能：API 接口，用于健康检查。
 * 路径：GET /api/health
 */
func (c *WebSocketController) HealthHandler(ctx *gin.Context) {
	ctx.JSON(http.StatusOK, gin.H{
		"status": "ok",
		"node":   c.NodeID,
		"port":   c.Port,
		"time":   time.Now().Format(time.RFC3339),
	})
}

/**
 * IsUserOnline
 * 功能：API 接口，检查指定用户当前是否在线（通过 Redis 查询）。
 * 路径：GET /api/check-user-online
 */
func (c *WebSocketController) IsUserOnline(ctx *gin.Context) {
	// 优先尝试查询参数，再尝试路径参数
	userID := ctx.Query("user_id")
	if userID == "" {
		userID = ctx.Param("user_id")
	}

	if userID == "" {
		ctx.JSON(http.StatusBadRequest, gin.H{
			"code":    -1,
			"status":  "error",
			"message": "缺少user_id参数",
		})
		return
	}

	// 使用与绑定逻辑一致的键名格式
	userClientsKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID)

	// 检查键是否存在并处理错误
	exists, err := c.RedisCli.Exists(c.RedisCtx, userClientsKey).Result()
	if err != nil {
		log.Printf("⚠️ 检查用户在线状态失败: %v | UserID=%s", err, userID)
		ctx.JSON(http.StatusInternalServerError, gin.H{
			"code":    -2,
			"status":  "error",
			"message": "服务内部错误",
			"result":  false,
		})
		return
	}

	online := exists > 0
	ctx.JSON(http.StatusOK, gin.H{
		"code":    0,
		"status":  "success",
		"message": "请求成功",
		"result":  online,
	})
}

/**
 * HandleWebSocket
 * 功能：处理 WebSocket 握手升级，并开启消息监听循环。
 * 路径：GET /ws
 */
func (c *WebSocketController) HandleWebSocket(ctx *gin.Context) {
	start := time.Now()
	clientIP := ctx.ClientIP()
	log.Printf("👤👤👤👤👤👤👤👤 客户端连接中: IP=%s", clientIP)

	conn, err := upgrader.Upgrade(ctx.Writer, ctx.Request, nil)
	if err != nil {
		log.Printf("⚠️ WebSocket升级失败: %v | ClientIP=%s", err, clientIP)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "无法升级为WebSocket连接"})
		return
	}
	defer conn.Close()

	clientID := utils.GenerateClientID(c.NodeID)
	log.Printf("✅ 客户端已连接: \n  ClientID=%s \n  IP=%s \n  Duration=%s",
		clientID, clientIP, time.Since(start))

	c.addClient(clientID, conn)
	defer c.removeClient(clientID)

	if err := conn.WriteJSON(gin.H{"clientId": clientID}); err != nil {
		log.Printf("⚠️ 发送客户端ID失败: %v | ClientID=%s", err, clientID)
		return
	}

	conn.SetPongHandler(func(string) error {
		conn.SetReadDeadline(time.Now().Add(60 * time.Second))
		return nil
	})

	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("❌❌❌❌❌❌❌❌ 连接意外断开: %v | ClientID=%s", err, clientID)
			} else {
				log.Printf("⚠️ 连接正常关闭: ClientID=%s", clientID)
			}
			break
		}

		if messageType == websocket.TextMessage {
			go c.handleClientMessage(clientID, message)
		}
	}
}

/**
 * handleClientMessage
 * 功能：处理从客户端接收到的消息，支持绑定、发消息、通话信令等操作。
 * @param senderID string 发送者的客户端ID
 * @param message []byte 原始消息体
 */
func (c *WebSocketController) handleClientMessage(senderID string, message []byte) {
	log.Printf("📥📥📥📥📥📥📥📥 收到客户端消息: \n  SenderID=%s \n  Size=%d bytes", senderID, len(message))

	var payload models.SendMessagePayload
	if err := json.Unmarshal(message, &payload); err == nil && payload.RequestType != "" {
		log.Printf("📦📦📦📦📦📦📦📦 解析JSON消息成功: Type=%s", payload.RequestType)

		// 处理unbind请求（清理旧连接）
		// 注意：unbind主要用于清理其他旧连接，不应该关闭当前连接
		// 如果前端需要完全断开，应该直接关闭WebSocket连接
		if payload.RequestType == "unbind" && payload.SenderUserID != "" {
			log.Printf("🔓🔓🔓🔓🔓🔓🔓🔓 处理解绑请求: \n  ClientID=%s \n  UserID=%s", senderID, payload.SenderUserID)

			// 清理该用户的所有旧连接（除了当前连接）
			// 这样可以确保旧连接被清理，但当前连接仍然可用
			c.cleanOldUserConnections(payload.SenderUserID, senderID)

			log.Printf("✅ 解绑完成: ClientID=%s | UserID=%s (已清理旧连接，当前连接保持)", senderID, payload.SenderUserID)
			return
		}

		if payload.RequestType == "bind" && payload.SenderUserID != "" {
			log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理绑定请求: \n  ClientID=%s \n  UserID=%s \n  UserType=%s \n  Token=%s \n  CleanOldConnections=%v",
				senderID, payload.SenderUserID, payload.UserType, maskToken(payload.Token), payload.CleanOldConnections)

			// 验证Token
			if payload.Token == "" {
				log.Printf("❌ 绑定失败: Token为空 | ClientID=%s", senderID)
				c.SendAuthResponse(senderID, "failed", "Token不能为空")
				return
			}

			// 根据用户类型选择验证方法
			userInfo, isValid := c.ValidateTokenByUserType(payload.Token, payload.UserType)
			if !isValid {
				log.Printf("❌ 绑定失败: Token验证失败 | ClientID=%s | UserType=%s", senderID, payload.UserType)
				c.SendAuthResponse(senderID, "failed", "Token验证失败，请重新登录")
				return
			}

			// 绑定用户（检查是否需要清理旧连接）
			cleanOldConnections := payload.CleanOldConnections
			if err := c.bindClientToUser(senderID, payload.SenderUserID, cleanOldConnections); err != nil {
				log.Printf("❌❌❌❌❌❌❌❌ 绑定失败: %v", err)
				c.SendAuthResponse(senderID, "failed", "绑定失败")
			} else {
				log.Printf("✅ 绑定成功: \n  ClientID=%s \n  UserID=%s \n  CleanOldConnections=%v", senderID, payload.SenderUserID, cleanOldConnections)

				// 保存客户端认证信息
				clientInfo := &models.ClientInfo{
					ClientID:      senderID,
					UserID:        payload.SenderUserID,
					UserType:      payload.UserType,
					Token:         payload.Token,
					IsAuth:        true,
					AuthTime:      time.Now(),
					LastCheckTime: time.Now(),
				}
				c.SaveClientInfo(senderID, clientInfo)

				// 发送认证成功响应
				c.SendAuthResponse(senderID, "success", "认证成功")

				// 启动定期Token检查
				go c.startTokenCheckTimer(senderID)

				log.Printf("🔐 客户端认证信息已保存: \n  ClientID=%s \n  UserInfo=%s", senderID, userInfo)
			}
			return
		}

		if payload.RequestType == "send_message" {
			if payload.TargetClientID != "" {
				log.Printf("📨📨📨📨📨📨📨📨 处理客户端发起的发送请求: \n  TargetID=%s \n  MsgType=%d",
					payload.TargetClientID, payload.MessageType)

				clientMsg := models.ClientReceivedMessage{
					SenderID:       senderID,
					ReceiverID:     payload.TargetClientID,
					SenderUserID:   payload.SenderUserID,
					ReceiverUserID: payload.ReceiverUserID,
					MessageType:    payload.MessageType,
					MessageContent: payload.MessageContent,
					CreatedAt:      time.Now().Format(time.RFC3339),
				}

				c.sendMessageToClient(payload.TargetClientID, clientMsg)
				return
			}

			if payload.ReceiverUserID != "" {
				log.Printf("📨📨📨📨📨📨📨📨 处理客户端发起的用户发送请求: \n  ReceiverUserID=%s \n  MsgType=%d",
					payload.ReceiverUserID, payload.MessageType)

				clientMsg := models.ClientReceivedMessage{
					SenderID:       senderID,
					ReceiverID:     "",
					SenderUserID:   payload.SenderUserID,
					ReceiverUserID: payload.ReceiverUserID,
					MessageType:    payload.MessageType,
					MessageContent: payload.MessageContent,
					CreatedAt:      time.Now().Format(time.RFC3339),
				}

				c.sendMessageToUser(payload.ReceiverUserID, clientMsg)
				return
			}
		}
	}

	var clientMsg models.ClientReceivedMessage
	if err := json.Unmarshal(message, &clientMsg); err == nil && clientMsg.ReceiverID != "" {
		log.Printf("📨📨📨📨📨📨📨📨 处理客户端封装消息: \n  TargetID=%s \n  MsgType=%d",
			clientMsg.ReceiverID, clientMsg.MessageType)

		clientMsg.SenderID = senderID
		c.sendMessageToClient(clientMsg.ReceiverID, clientMsg)
		return
	}

	// 处理通话信令
	var callSignal models.CallSignal
	if err := json.Unmarshal(message, &callSignal); err == nil && callSignal.CallStatus != "" {
		// 获取发送者用户ID
		senderUserID, _ := c.getUserIDByClientID(senderID)

		log.Printf("📞📞 处理通话信令: \n  CallStatus=%s \n  CallID=%s \n  CallType=%d \n  SenderUserID=%s \n  CallerID=%s \n  CalleeID=%s",
			callSignal.CallStatus, callSignal.CallID, callSignal.CallType, senderUserID, callSignal.CallerID, callSignal.CalleeID)

		// 处理不同类型的通话动作
		switch callSignal.CallStatus {
		case "invite":
			c.handleCallInvite(senderID, senderUserID, callSignal)
		case "accepted":
			c.handleCallAccept(senderID, senderUserID, callSignal)
		case "rejected":
			c.handleCallReject(senderID, senderUserID, callSignal)
		case "ended":
			c.handleCallEnd(senderID, senderUserID, callSignal)
		case "offer":
			c.handleCallOffer(senderID, senderUserID, callSignal)
		case "answer":
			c.handleCallAnswer(senderID, senderUserID, callSignal)
		case "candidate":
			c.handleCallCandidate(senderID, senderUserID, callSignal)
		case "hangup":
			c.handleCallHangup(senderID, senderUserID, callSignal)
		case "disconnected":
			c.handleCallDisconnected(senderID, senderUserID, callSignal)
		case "terminated":
			c.handleCallTerminated(senderID, senderUserID, callSignal)
		case "no-answer":
			c.handleCallNoAnswer(senderID, senderUserID, callSignal)
		case "busy":
			c.handleCallBusy(senderID, senderUserID, callSignal)
		case "failed":
			c.handleCallFailed(senderID, senderUserID, callSignal)
		default:
			log.Printf("⚠️ 未知的通话动作: %s", callSignal.CallStatus)
		}
		return
	}

	log.Printf("⚠️ 无法识别的消息格式: \n  Size=%d bytes \n  Message=%s", len(message), string(message))
}

/**
 * handleCallInvite
 * 功能：处理WebRTC通话邀请信令。
 */
func (c *WebSocketController) handleCallInvite(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理通话邀请: \n  CallID=%s \n  CallType=%d \n  From=%s \n  To=%s",
		signal.CallID, signal.CallType, senderUserID, signal.CalleeID)

	// 创建通话消息
	callMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "invite",
	}

	// 发送邀请给被叫方
	if err := c.sendMessageToUser(signal.CalleeID, callMsg); err != nil {
		log.Printf("❌ 发送通话邀请失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallAccept
 * 功能：处理被叫方接受通话的信令。
 */
func (c *WebSocketController) handleCallAccept(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("✅ 处理通话接受: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)

	// 创建通话接受消息
	acceptMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CallerID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "accepted",
	}

	// 发送接受消息给主叫方
	if err := c.sendMessageToUser(signal.CallerID, acceptMsg); err != nil {
		log.Printf("❌ 发送通话接受失败: %v | CallerID=%s", err, signal.CallerID)
	}
}

/**
 * handleCallReject
 * 功能：处理被叫方拒绝通话的信令。
 */
func (c *WebSocketController) handleCallReject(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("❌ 处理通话拒绝: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)

	// 创建通话拒绝消息
	rejectMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CallerID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "rejected",
	}

	// 发送拒绝消息给主叫方
	if err := c.sendMessageToUser(signal.CallerID, rejectMsg); err != nil {
		log.Printf("❌ 发送通话拒绝失败: %v | CallerID=%s", err, signal.CallerID)
	}
}

/**
 * handleCallEnd
 * 功能：处理通话结束/挂断信令。
 */
func (c *WebSocketController) handleCallEnd(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理通话结束: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建通话结束消息
	endMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "ended",
	}

	// 发送结束消息给对端
	if err := c.sendMessageToUser(signal.CalleeID, endMsg); err != nil {
		log.Printf("❌ 发送通话结束失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallOffer
 * 功能：处理 WebRTC 的 Offer 信令（SDP交换）。
 */
func (c *WebSocketController) handleCallOffer(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理Offer信令: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建Offer消息
	offerMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "offer",
	}

	// 发送Offer给对端
	if err := c.sendMessageToUser(signal.CalleeID, offerMsg); err != nil {
		log.Printf("❌ 发送Offer失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallAnswer
 * 功能：处理 WebRTC 的 Answer 信令（SDP交换）。
 */
func (c *WebSocketController) handleCallAnswer(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理Answer信令: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)

	// 创建Answer消息
	answerMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CallerID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "answer",
	}

	// 发送Answer给对端
	if err := c.sendMessageToUser(signal.CallerID, answerMsg); err != nil {
		log.Printf("❌ 发送Answer失败: %v | CallerID=%s", err, signal.CallerID)
	}
}

/**
 * handleCallCandidate
 * 功能：处理 ICE Candidate 信令（网络协商）。
 */
func (c *WebSocketController) handleCallCandidate(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📶 处理通话候选: \n  CallID=%s  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建候选消息
	candidateMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "candidate",
	}

	// 发送候选消息给对端
	if err := c.sendMessageToUser(signal.CalleeID, candidateMsg); err != nil {
		log.Printf("❌ 发送候选失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallHangup
 * 功能：处理主动挂断通话。
 */
func (c *WebSocketController) handleCallHangup(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理通话挂断: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建挂断消息
	hangupMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "hangup",
	}

	// 发送挂断消息给对端
	if err := c.sendMessageToUser(signal.CalleeID, hangupMsg); err != nil {
		log.Printf("❌ 发送通话挂断失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallDisconnected
 * 功能：处理连接意外断开。
 */
func (c *WebSocketController) handleCallDisconnected(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理通话掉线: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建掉线消息
	disconnectedMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "disconnected",
	}

	// 发送掉线消息给对端
	if err := c.sendMessageToUser(signal.CalleeID, disconnectedMsg); err != nil {
		log.Printf("❌ 发送通话掉线失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallTerminated
 * 功能：处理通话终止信号。
 */
func (c *WebSocketController) handleCallTerminated(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理通话终止: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建终止消息
	terminatedMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "terminated",
	}

	// 发送终止消息给对端
	if err := c.sendMessageToUser(signal.CalleeID, terminatedMsg); err != nil {
		log.Printf("❌ 发送通话终止失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * handleCallNoAnswer
 * 功能：处理无人接听状态。
 */
func (c *WebSocketController) handleCallNoAnswer(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理无人接听: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)

	// 创建无人接听消息
	noAnswerMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CallerID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "no-answer",
	}

	// 发送无人接听消息
	if err := c.sendMessageToUser(signal.CallerID, noAnswerMsg); err != nil {
		log.Printf("❌ 发送无人接听失败: %v | CallerID=%s", err, signal.CallerID)
	}
}

/**
 * handleCallBusy
 * 功能：处理忙线状态。
 */
func (c *WebSocketController) handleCallBusy(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理忙线状态: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)

	// 创建忙线消息
	busyMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CallerID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "busy",
	}

	// 发送忙线消息
	if err := c.sendMessageToUser(signal.CallerID, busyMsg); err != nil {
		log.Printf("❌ 发送忙线状态失败: %v | CallerID=%s", err, signal.CallerID)
	}
}

/**
 * handleCallFailed
 * 功能：处理通话建立失败状态。
 */
func (c *WebSocketController) handleCallFailed(senderID, senderUserID string, signal models.CallSignal) {
	log.Printf("📞 处理通话失败: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)

	// 创建失败消息
	failedMsg := models.ClientReceivedMessage{
		SenderID:       senderID,
		ReceiverID:     "",
		SenderUserID:   senderUserID,
		ReceiverUserID: signal.CalleeID,
		MessageType:    signal.CallType,
		MessageContent: signal.Data,
		CreatedAt:      time.Now().Format(time.RFC3339),
		CallID:         signal.CallID,
		CallStatus:     "failed",
	}

	// 发送失败消息
	if err := c.sendMessageToUser(signal.CalleeID, failedMsg); err != nil {
		log.Printf("❌ 发送通话失败消息失败: %v | CalleeID=%s", err, signal.CalleeID)
	}
}

/**
 * SendMessageHandler
 * 功能：API 接口，用于服务器端主动发送消息（HTTP -> WebSocket）。
 * 路径：POST /api/send
 */
func (c *WebSocketController) SendMessageHandler(ctx *gin.Context) {
	start := time.Now()

	var payload models.SendMessagePayload
	if err := ctx.ShouldBindJSON(&payload); err != nil {
		log.Printf("⚠️ 无效的JSON请求格式: %v", err)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "无效的JSON格式"})
		return
	}

	if payload.RequestType == "" || (payload.TargetClientID == "" && payload.ReceiverUserID == "") ||
		payload.SenderUserID == "" || payload.ReceiverUserID == "" {
		log.Printf("⚠️ 缺少必要参数: \n  request_type=%s \n  target_client_id=%s \n  receiver_user_id=%s \n  sender_user_id=%s \n  receiver_user_id=%s",
			payload.RequestType, payload.TargetClientID, payload.ReceiverUserID, payload.SenderUserID, payload.ReceiverUserID)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "缺少必要参数"})
		return
	}

	if payload.RequestType == "send_message" {
		log.Printf("📤📤📤📤📤📤📤📤 处理API发送请求: \n  SenderUser=%s \n  MsgType=%d",
			payload.SenderUserID, payload.MessageType)

		clientMsg := models.ClientReceivedMessage{
			SenderID:       "system",
			RoomID:         payload.RoomId,
			ReceiverID:     payload.TargetClientID,
			SenderUserID:   payload.SenderUserID,
			ReceiverUserID: payload.ReceiverUserID,
			MessageType:    payload.MessageType,
			MessageContent: payload.MessageContent,
			CreatedAt:      time.Now().Format(time.RFC3339),
		}

		var result string
		var sendErr error

		if payload.TargetClientID != "" {
			log.Printf("🎯🎯🎯🎯🎯🎯🎯🎯 目标类型: ClientID | Target=%s", payload.TargetClientID)
			sendErr = c.sendMessageToClient(payload.TargetClientID, clientMsg)
			result = fmt.Sprintf("消息已发送到ClientID: %s", payload.TargetClientID)
		} else if payload.ReceiverUserID != "" {
			log.Printf("🎯🎯🎯🎯🎯🎯🎯🎯 目标类型: UserID | ReceiverUser=%s", payload.ReceiverUserID)
			sendErr = c.sendMessageToUser(payload.ReceiverUserID, clientMsg)
			result = fmt.Sprintf("消息已发送到UserID: %s", payload.ReceiverUserID)
		} else {
			ctx.JSON(http.StatusBadRequest, gin.H{"error": "必须指定目标客户端ID或用户ID"})
			return
		}

		if sendErr != nil {
			ctx.JSON(http.StatusInternalServerError, gin.H{"error": sendErr.Error()})
			return
		}

		ctx.JSON(http.StatusOK, gin.H{"status": "success", "message": result})
		log.Printf("✅ API请求完成: Duration=%s", time.Since(start))
		return
	}

	ctx.JSON(http.StatusBadRequest, gin.H{"error": "不支持的request_type"})
}

/**
 * BindHandler
 * 功能：API 接口，用于强制绑定 ClientID 和 UserID。
 * 路径：POST /api/bind
 */
func (c *WebSocketController) BindHandler(ctx *gin.Context) {
	start := time.Now()

	var req models.BindRequest
	if err := ctx.ShouldBindJSON(&req); err != nil {
		log.Printf("⚠️ 无效的绑定请求格式: %v", err)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "无效的JSON格式"})
		return
	}

	if req.UserID == "" || req.ClientID == "" {
		log.Printf("⚠️ 缺少必要参数: \n  user_id=%s \n  client_id=%s", req.UserID, req.ClientID)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "user_id和client_id不能为空"})
		return
	}

	log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理用户绑定请求: \n  UserID=%s \n  ClientID=%s", req.UserID, req.ClientID)

	// API绑定请求默认清理旧连接
	if err := c.bindClientToUser(req.ClientID, req.UserID, true); err != nil {
		log.Printf("❌❌❌❌❌❌❌❌ 绑定失败: %v", err)
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
		return
	}

	ctx.JSON(http.StatusOK, gin.H{
		"status":  "success",
		"message": fmt.Sprintf("绑定成功: ClientID %s -> UserID %s", req.ClientID, req.UserID),
	})

	log.Printf("✅ 绑定请求完成: \n  UserID=%s \n  ClientID=%s \n  Duration=%s",
		req.UserID, req.ClientID, time.Since(start))
}

/**
 * SendToUserHandler
 * 功能：API 接口，向指定用户的所有在线设备发送消息，并持久化到数据库。
 * 路径：POST /api/send-to-user
 */
// SendToUserHandler 通过用户ID发送消息处理器（支持通话信令）
func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) {
	start := time.Now()

	var payload models.SendToUserPayload
	if err := ctx.ShouldBindJSON(&payload); err != nil {
		log.Printf("⚠️ 无效的JSON请求格式: %v", err)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "无效的JSON格式"})
		return
	}

	if payload.ReceiverUserID == "" || payload.SenderUserID == "" {
		log.Printf("⚠️ 缺少必要参数: \n  receiver_user_id=%s \n  sender_user_id=%s",
			payload.ReceiverUserID, payload.SenderUserID)
		ctx.JSON(http.StatusBadRequest, gin.H{"error": "缺少接收方或发送方用户ID"})
		return
	}

	log.Printf("📤📤📤📤📤📤📤📤 处理API用户发送请求: \n  SenderUser=%s → ReceiverUser=%s \n  MsgType=%d",
		payload.SenderUserID, payload.ReceiverUserID, payload.MessageType)

	// 创建消息记录
	chatMsg := models.XkChatMessage{
		RoomId:         payload.RoomId,
		SenderUserID:   payload.SenderUserID,
		ReceiverUserID: payload.ReceiverUserID,
		Duration:       payload.Duration,
		MessageType:    payload.MessageType,
		MessageContent: payload.MessageContent,
		CallID:         payload.CallID,
		CallStatus:     payload.CallStatus,
	}

	var savedID = 0

	// 保存到数据库
	if err := c.DB.Create(&chatMsg).Error; err != nil {
		log.Printf("❌ 保存消息到数据库失败: %v", err)
	} else {
		// 获取消息id
		log.Printf("💾 消息保存到数据库成功: ID=%d", chatMsg.ID)
		savedID = int(chatMsg.ID)
	}

	// 创建基础消息结构
	clientMsg := models.ClientReceivedMessage{
		SenderID:       "system",
		ReceiverID:     "",
		ID:             savedID,
		RoomID:         payload.RoomId,
		SenderUserID:   payload.SenderUserID,
		ReceiverUserID: payload.ReceiverUserID,
		Duration:       payload.Duration,
		MessageType:    payload.MessageType,
		MessageContent: payload.MessageContent,
		CreatedAt:      time.Now().Format(time.RFC3339),
	}

	// 如果是通话信令消息，添加通话相关字段
	if payload.MessageType == 6 || payload.MessageType == 7 {
		clientMsg.CallStatus = payload.CallStatus
		clientMsg.CallID = payload.CallID

		log.Printf("📞📞 处理通话信令消息: \n  CallStatus=%s \n  CallID=%s",
			payload.CallStatus, payload.CallID)
	}

	// 发送消息给目标用户
	if err := c.sendMessageToUser(payload.ReceiverUserID, clientMsg); err != nil {
		log.Printf("❌ 发送消息失败: %v | ReceiverUserID=%s", err, payload.ReceiverUserID)
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
		return
	}

	ctx.JSON(http.StatusOK, gin.H{
		"status":  "success",
		"message": fmt.Sprintf("消息已发送到用户: %s", payload.ReceiverUserID),
	})

	log.Printf("✅ API用户发送请求完成: Duration=%s", time.Since(start))
}

/**
 * GetMessagesHandler
 * 功能：API 接口，分页查询历史聊天记录。
 * 路径：GET /api/messages
 */
func (c *WebSocketController) GetMessagesHandler(ctx *gin.Context) {
	// 解析查询参数
	roomId := ctx.Query("room_id")
	senderUserId := ctx.Query("sender_user_id")
	receiverUserId := ctx.Query("receiver_user_id")

	page, _ := strconv.Atoi(ctx.DefaultQuery("page", "1"))
	pageSize, _ := strconv.Atoi(ctx.DefaultQuery("page_size", "15"))

	if page < 1 {
		page = 1
	}
	if pageSize < 1 || pageSize > 100 {
		pageSize = 15
	}

	offset := (page - 1) * pageSize

	// 构建查询
	query := c.DB.Model(&models.XkChatMessage{})
	if roomId != "" {
		query = query.Where("room_id = ?", roomId)
	}
	if senderUserId != "" {
		query = query.Where("sender_user_id = ?", senderUserId)
	}
	if receiverUserId != "" {
		query = query.Where("receiver_user_id = ?", receiverUserId)
	}

	// 获取总数
	var total int64
	query.Count(&total)

	// 获取分页数据
	var messages []models.XkChatMessage
	if err := query.Order("created_at DESC").Offset(offset).Limit(pageSize).Find(&messages).Error; err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "获取消息失败"})
		return
	}

	ctx.JSON(http.StatusOK, gin.H{
		"status":  "success",
		"data":    messages,
		"total":   total,
		"page":    page,
		"hasNext": total > int64(offset+pageSize),
	})
}

/**
 * SyncMessagesHandler
 * 功能：API 接口，获取全部聊天记录（用于同步）。
 * 路径：GET /api/messages/sync
 */
func (c *WebSocketController) SyncMessagesHandler(ctx *gin.Context) {
	roomId := ctx.Query("room_id")
	senderUserId := ctx.Query("sender_user_id")
	receiverUserId := ctx.Query("receiver_user_id")

	query := c.DB.Model(&models.XkChatMessage{})
	if roomId != "" {
		query = query.Where("room_id = ?", roomId)
	}
	if senderUserId != "" {
		query = query.Where("sender_user_id = ?", senderUserId)
	}
	if receiverUserId != "" {
		query = query.Where("receiver_user_id = ?", receiverUserId)
	}

	var messages []models.XkChatMessage
	if err := query.Order("created_at ASC").Find(&messages).Error; err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "获取消息失败"})
		return
	}

	ctx.JSON(http.StatusOK, gin.H{
		"status": "success",
		"data":   messages,
	})
}

/**
 * addClient
 * 功能：将新的 WebSocket 连接添加到本地连接池。
 * 注意：使用 Mutex 锁保证线程安全。
 */
func (c *WebSocketController) addClient(clientID string, conn *websocket.Conn) {
	c.ClientsMux.Lock()
	defer c.ClientsMux.Unlock()
	c.Clients[clientID] = conn
	log.Printf("📊📊📊📊📊📊📊📊 当前连接数: %d", len(c.Clients))
}

/**
 * removeClient
 * 功能：从本地连接池移除客户端，并清理 Redis 中的绑定关系。
 */
func (c *WebSocketController) removeClient(clientID string) {
	c.ClientsMux.Lock()
	defer c.ClientsMux.Unlock()

	if _, exists := c.Clients[clientID]; exists {
		delete(c.Clients, clientID)
		log.Printf("🔌🔌🔌🔌🔌🔌🔌🔌 客户端已断开: ClientID=%s | 剩余连接数: %d", clientID, len(c.Clients))

		// 清理用户绑定关系
		c.cleanupUserBinding(clientID)

		// 清理客户端认证信息
		c.RemoveClientInfo(clientID)
		log.Printf("🧹 已清理客户端认证信息: ClientID=%s", clientID)
	}
}

/**
 * cleanupUserBinding
 * 功能：从 Redis 中清除用户与客户端ID的映射关系。
 */
func (c *WebSocketController) cleanupUserBinding(clientID string) {
	// 获取用户ID
	userID, err := c.getUserIDByClientID(clientID)
	if err != nil {
		return
	}

	// 从用户-客户端映射中移除
	userClientsKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID)
	c.RedisCli.SRem(c.RedisCtx, userClientsKey, clientID)

	// 从客户端-用户映射中移除
	clientUserKey := fmt.Sprintf("%s:%s", models.ClientUserKey, clientID)
	c.RedisCli.Del(c.RedisCtx, clientUserKey)

	log.Printf("🧹🧹🧹🧹🧹🧹🧹🧹 清理绑定关系: ClientID=%s | UserID=%s", clientID, userID)
}

/**
 * cleanOldUserConnections
 * 功能：清理用户的所有旧连接（除了当前clientID）
 * @param userID string 用户ID
 * @param currentClientID string 当前客户端ID（保留此连接）
 */
func (c *WebSocketController) cleanOldUserConnections(userID, currentClientID string) {
	// 获取该用户的所有客户端ID
	userClientsKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID)
	clientIDs, err := c.RedisCli.SMembers(c.RedisCtx, userClientsKey).Result()
	if err != nil {
		log.Printf("⚠️ 获取用户客户端列表失败: %v | UserID=%s", err, userID)
		return
	}

	if len(clientIDs) == 0 {
		log.Printf("ℹ️ 用户无旧连接需要清理: UserID=%s", userID)
		return
	}

	log.Printf("🧹🧹🧹🧹🧹🧹🧹🧹 开始清理用户旧连接: \n  UserID=%s \n  当前ClientID=%s \n  旧连接数=%d", userID, currentClientID, len(clientIDs))

	cleanedCount := 0
	for _, oldClientID := range clientIDs {
		// 跳过当前客户端ID
		if oldClientID == currentClientID {
			continue
		}

		// 检查旧连接是否还在本节点
		c.ClientsMux.RLock()
		conn, exists := c.Clients[oldClientID]
		c.ClientsMux.RUnlock()

		if exists {
			// 如果连接还在本节点，先关闭WebSocket连接
			log.Printf("🔌 关闭旧连接: ClientID=%s | UserID=%s", oldClientID, userID)
			if conn != nil {
				conn.Close()
			}
			// 然后调用removeClient清理（会同时清理Redis映射和本地连接）
			c.removeClient(oldClientID)
		} else {
			// 如果连接不在本节点，只清理Redis映射
			log.Printf("🧹 清理Redis映射: ClientID=%s | UserID=%s", oldClientID, userID)
			// 从用户-客户端映射中移除
			c.RedisCli.SRem(c.RedisCtx, userClientsKey, oldClientID)
			// 从客户端-用户映射中移除
			clientUserKey := fmt.Sprintf("%s:%s", models.ClientUserKey, oldClientID)
			c.RedisCli.Del(c.RedisCtx, clientUserKey)
		}

		cleanedCount++
	}

	// 清空用户的所有客户端映射（稍后会在bindClientToUser中重新添加当前clientID）
	if cleanedCount > 0 {
		// 先移除当前clientID（如果存在），然后清空整个集合
		c.RedisCli.SRem(c.RedisCtx, userClientsKey, currentClientID)
		c.RedisCli.Del(c.RedisCtx, userClientsKey)
		log.Printf("✅ 清理完成: \n  UserID=%s \n  清理连接数=%d", userID, cleanedCount)
	}
}

/**
 * bindClientToUser
 * 功能：在 Redis 中建立 UserID 和 ClientID 的双向映射。
 * 映射1：client_user_mapping:clientID -> userID (string)
 * 映射2：user_client_mapping:userID -> [clientID1, clientID2] (set)
 * @param clientID string 客户端ID
 * @param userID string 用户ID
 * @param cleanOldConnections bool 是否清理该用户之前的旧连接（默认false）
 */
func (c *WebSocketController) bindClientToUser(clientID, userID string, cleanOldConnections bool) error {
	// 如果需要清理旧连接，先清理该用户的所有旧连接
	if cleanOldConnections {
		c.cleanOldUserConnections(userID, clientID)
	}

	// 设置客户端->用户映射
	clientUserKey := fmt.Sprintf("%s:%s", models.ClientUserKey, clientID)
	if err := c.RedisCli.Set(c.RedisCtx, clientUserKey, userID, 0).Err(); err != nil {
		return fmt.Errorf("设置客户端用户映射失败: %v", err)
	}

	// 添加用户->客户端集合映射
	userClientsKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID)
	if err := c.RedisCli.SAdd(c.RedisCtx, userClientsKey, clientID).Err(); err != nil {
		return fmt.Errorf("添加用户客户端集合失败: %v", err)
	}

	log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 绑定成功: \n  ClientID=%s \n  UserID=%s \n  CleanOldConnections=%v", clientID, userID, cleanOldConnections)
	return nil
}

// 根据客户端ID获取用户ID
func (c *WebSocketController) getUserIDByClientID(clientID string) (string, error) {
	clientUserKey := fmt.Sprintf("%s:%s", models.ClientUserKey, clientID)
	userID, err := c.RedisCli.Get(c.RedisCtx, clientUserKey).Result()
	if err != nil {
		return "", fmt.Errorf("获取用户ID失败: %v", err)
	}
	return userID, nil
}

// 根据用户ID获取客户端ID列表
func (c *WebSocketController) getClientIDsByUserID(userID string) ([]string, error) {
	userClientsKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID)
	clientIDs, err := c.RedisCli.SMembers(c.RedisCtx, userClientsKey).Result()
	if err != nil {
		return nil, fmt.Errorf("获取客户端ID列表失败: %v", err)
	}
	return clientIDs, nil
}

/**
 * sendMessageToClient
 * 功能：向指定 ClientID 发送消息。
 * 逻辑：如果 Client 在本节点，直接通过 WebSocket 发送；如果不在，尝试通过 Redis 广播转发。
 */
func (c *WebSocketController) sendMessageToClient(clientID string, message models.ClientReceivedMessage) error {
	c.ClientsMux.RLock()
	conn, exists := c.Clients[clientID]
	c.ClientsMux.RUnlock()

	if !exists {
		// 尝试通过Redis转发到其他节点
		return c.forwardMessageToOtherNodes(clientID, message)
	}

	c.WriteMutex.Lock()
	defer c.WriteMutex.Unlock()

	if err := conn.WriteJSON(message); err != nil {
		log.Printf("❌ 发送消息失败: %v | ClientID=%s", err, clientID)
		c.removeClient(clientID)
		return err
	}

	log.Printf("✅ 消息已发送: \n  ClientID=%s \n  MsgType=%d \n  Content=%s",
		clientID, message.MessageType, message.MessageContent)
	return nil
}

/**
 * sendMessageToUser
 * 功能：向指定用户的所有设备发送消息。
 */
func (c *WebSocketController) sendMessageToUser(userID string, message models.ClientReceivedMessage) error {
	clientIDs, err := c.getClientIDsByUserID(userID)
	if err != nil {
		log.Printf("⚠️ 获取用户客户端列表失败: %v | UserID=%s", err, userID)
		return err
	}

	if len(clientIDs) == 0 {
		log.Printf("⚠️ 用户无在线客户端: UserID=%s", userID)
		//return fmt.Errorf("用户 %s 无在线客户端", userID)
	} else {
		log.Printf("📤📤📤📤📤📤📤📤 发送消息给用户: \n  UserID=%s \n  ClientCount=%d \n  MsgType=%d",
			userID, len(clientIDs), message.MessageType)

		var lastErr error
		successCount := 0

		for _, clientID := range clientIDs {
			if err := c.sendMessageToClient(clientID, message); err != nil {
				log.Printf("⚠️ 发送到客户端失败: %v | ClientID=%s", err, clientID)
				lastErr = err
			} else {
				successCount++
			}
		}

		if successCount == 0 {
			return fmt.Errorf("所有客户端发送失败，最后错误: %v", lastErr)
		}

		log.Printf("✅ 用户消息发送完成: \n  UserID=%s \n  成功=%d/%d", userID, successCount, len(clientIDs))
	}
	return nil
}

/**
 * forwardMessageToOtherNodes
 * 功能：当目标客户端不在本机时，通过 Redis Pub/Sub 将消息广播到集群其他节点。
 */
func (c *WebSocketController) forwardMessageToOtherNodes(clientID string, message models.ClientReceivedMessage) error {
	redisMsg := models.RedisMessage{
		SenderNodeID: c.NodeID,
		ClientID:     clientID,
		Message:      "",
	}

	msgBytes, err := json.Marshal(message)
	if err != nil {
		return fmt.Errorf("序列化消息失败: %v", err)
	}
	redisMsg.Message = string(msgBytes)

	redisMsgBytes, err := json.Marshal(redisMsg)
	if err != nil {
		return fmt.Errorf("序列化Redis消息失败: %v", err)
	}

	channel := fmt.Sprintf("websocket:forward:%s", clientID)
	if err := c.RedisCli.Publish(c.RedisCtx, channel, redisMsgBytes).Err(); err != nil {
		return fmt.Errorf("发布Redis消息失败: %v", err)
	}

	log.Printf("📡📡📡📡📡📡📡📡 消息已转发到其他节点: \n  ClientID=%s \n  Channel=%s", clientID, channel)
	return nil
}

/**
 * SubscribeToRedis
 * 功能：订阅 Redis 广播频道，监听来自其他节点的消息转发请求。
 */
func (c *WebSocketController) SubscribeToRedis() {
	go func() {
		pattern := "websocket:forward:*"
		pubsub := c.RedisCli.PSubscribe(c.RedisCtx, pattern)
		defer pubsub.Close()

		log.Printf("📡📡📡📡📡📡📡📡 Redis消息监听已启动: Pattern=%s", pattern)

		for msg := range pubsub.Channel() {
			var redisMsg models.RedisMessage
			if err := json.Unmarshal([]byte(msg.Payload), &redisMsg); err != nil {
				log.Printf("⚠️ 解析Redis消息失败: %v", err)
				continue
			}

			// 忽略自己发送的消息
			if redisMsg.SenderNodeID == c.NodeID {
				continue
			}

			// 检查目标客户端是否在本节点
			c.ClientsMux.RLock()
			conn, exists := c.Clients[redisMsg.ClientID]
			c.ClientsMux.RUnlock()

			if !exists {
				continue
			}

			var clientMsg models.ClientReceivedMessage
			if err := json.Unmarshal([]byte(redisMsg.Message), &clientMsg); err != nil {
				log.Printf("⚠️ 解析客户端消息失败: %v", err)
				continue
			}

			c.WriteMutex.Lock()
			if err := conn.WriteJSON(clientMsg); err != nil {
				log.Printf("❌ 转发消息失败: %v | ClientID=%s", err, redisMsg.ClientID)
				c.removeClient(redisMsg.ClientID)
			} else {
				log.Printf("✅ 转发消息成功: \n  ClientID=%s \n  FromNode=%s",
					redisMsg.ClientID, redisMsg.SenderNodeID)
			}
			c.WriteMutex.Unlock()
		}
	}()
}

/**
 * OpenAPIAuthMiddleware
 * 功能：Open API HMAC 签名验证中间件
 * 用于保护 /open-api 分组下的接口
 */
func (c *WebSocketController) OpenAPIAuthMiddleware() gin.HandlerFunc {
	return func(ctx *gin.Context) {
		// 1. 获取请求头中的签名信息
		timestamp := ctx.GetHeader("X-Timestamp")
		signature := ctx.GetHeader("X-Signature")

		if timestamp == "" || signature == "" {
			ctx.JSON(http.StatusUnauthorized, gin.H{"error": "缺少签名信息"})
			ctx.Abort()
			return
		}

		// 2. 验证时间戳（防重放攻击，允许5分钟误差）
		ts, err := strconv.ParseInt(timestamp, 10, 64)
		if err != nil {
			ctx.JSON(http.StatusUnauthorized, gin.H{"error": "无效的时间戳"})
			ctx.Abort()
			return
		}

		tolerance := int64(utils.GetEnvInt("OPEN_API_TIMESTAMP_TOLERANCE", 300))
		now := time.Now().Unix()
		diff := now - ts
		if diff < 0 {
			diff = -diff
		}
		if diff > tolerance {
			ctx.JSON(http.StatusUnauthorized, gin.H{"error": "请求已过期"})
			ctx.Abort()
			return
		}

		// 3. 读取请求体
		body, err := io.ReadAll(ctx.Request.Body)
		if err != nil {
			ctx.JSON(http.StatusBadRequest, gin.H{"error": "读取请求体失败"})
			ctx.Abort()
			return
		}
		// 重新设置请求体，供后续处理器使用
		ctx.Request.Body = io.NopCloser(bytes.NewBuffer(body))

		// 4. 计算签名
		secretKey := utils.GetEnv("OPEN_API_SECRET_KEY", "")
		if secretKey == "" {
			log.Printf("⚠️ OPEN_API_SECRET_KEY 未配置")
			ctx.JSON(http.StatusInternalServerError, gin.H{"error": "服务配置错误"})
			ctx.Abort()
			return
		}

		// 签名算法: HMAC-SHA256(secret, timestamp + body)
		message := timestamp + string(body)
		h := hmac.New(sha256.New, []byte(secretKey))
		h.Write([]byte(message))
		expectedSignature := hex.EncodeToString(h.Sum(nil))

		// 5. 验证签名
		if !hmac.Equal([]byte(signature), []byte(expectedSignature)) {
			log.Printf("❌ Open API 签名验证失败: expected=%s, got=%s", expectedSignature, signature)
			ctx.JSON(http.StatusUnauthorized, gin.H{"error": "签名验证失败"})
			ctx.Abort()
			return
		}

		log.Printf("✅ Open API 签名验证通过")
		ctx.Next()
	}
}

/**
 * APITokenAuthMiddleware
 * 功能：API Token 验证中间件
 * 用于客户端调用的接口，验证用户 Token
 * Token 从请求头 Authorization 获取
 * 用户类型从请求体 sender_user_id 推断（doctor- 开头为 doctor，否则为 user）
 */
func (c *WebSocketController) APITokenAuthMiddleware() gin.HandlerFunc {
	return func(ctx *gin.Context) {
		// 1. 获取 Token（从 Header 的 Authorization 字段）
		token := ctx.GetHeader("Authorization")
		if token == "" {
			token = ctx.GetHeader("authorization") // 兼容小写
		}

		// 移除 Bearer 前缀
		if len(token) > 7 && (token[:7] == "Bearer " || token[:7] == "bearer ") {
			token = token[7:]
		}

		if token == "" {
			log.Printf("❌ API Token 验证失败: Token 为空")
			ctx.JSON(http.StatusUnauthorized, gin.H{"error": "缺少认证信息"})
			ctx.Abort()
			return
		}

		// 2. 读取请求体获取 sender_user_id 以推断用户类型
		body, err := io.ReadAll(ctx.Request.Body)
		if err != nil {
			log.Printf("❌ API Token 验证失败: 读取请求体失败")
			ctx.JSON(http.StatusBadRequest, gin.H{"error": "读取请求体失败"})
			ctx.Abort()
			return
		}
		// 重新设置请求体，供后续处理器使用
		ctx.Request.Body = io.NopCloser(bytes.NewBuffer(body))

		// 3. 从请求体解析 sender_user_id
		var payload struct {
			SenderUserID string `json:"sender_user_id"`
		}
		userType := "user" // 默认为 user
		if err := json.Unmarshal(body, &payload); err == nil && payload.SenderUserID != "" {
			// 根据 sender_user_id 前缀判断用户类型
			if len(payload.SenderUserID) > 7 && payload.SenderUserID[:7] == "doctor-" {
				userType = "doctor"
			}
		}

		log.Printf("🔐 API Token 验证: UserType=%s | SenderUserID=%s", userType, payload.SenderUserID)

		// 4. 验证 Token
		userInfo, isValid := c.ValidateTokenByUserType(token, userType)
		if !isValid {
			log.Printf("❌ API Token 验证失败: Token 无效 | UserType=%s", userType)
			ctx.JSON(http.StatusUnauthorized, gin.H{"error": "Token 验证失败，请重新登录"})
			ctx.Abort()
			return
		}

		// 5. 将用户信息存入上下文，供后续处理器使用
		ctx.Set("userInfo", userInfo)
		ctx.Set("userType", userType)

		log.Printf("✅ API Token 验证通过 | UserType=%s", userType)
		ctx.Next()
	}
}