message-server/wsClient/room.go

// 房间管理模块 -- 负责创建、管理聊天房间，处理客户端的加入、离开和消息转发
package wsClient

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"math/rand"
	"net/http"
	"net/url"
	"strings"
	"sync"

	"github.com/gorilla/websocket"
	"github.com/rs/zerolog/log"
)

// 全局房间 map，存储所有创建的房间
// key: 房间名称，value: 房间对象指针
var rooms = make(map[string]*Room)

// Room 表示一个聊天房间
type Room struct {
	// holds all current clientConns in the room
	// 保存房间中所有当前活跃的客户端连接
	clientConns map[*Client]bool

	// join is a channel for all clients wishing to join the room
	// 用于接收希望加入房间的客户端。新客户端通过这个 channel 请求加入
	join chan *Client

	// leave is a channel for all clients wishing to leave the room
	// 用于接收希望离开房间的客户端。断开连接的客户端通过这个 channel 通知离开
	leave chan *Client

	// forward is a channel that holds incoming messages that should be forwarded to the other clients.
	// 用于接收需要转发给其他客户端的消息
	// 所有向此房间发送的消息都会通过这个 channel 进行分发
	forward chan *message

	// name 房间名称，用作Redis存储的键
	name string
}

// message 包含要转发的消息内容和发送者信息
type message struct {
	content []byte  // 消息内容
	sender  *Client // 发送者 -- // 用处：1.用于在转发时排除发送者自己
}

// RoomParams 表示从客户端 WebSocket子协议中解析出的房间参数
type RoomParams struct {
	RoomName  string `json:"roomName,omitempty"`
	UserId    int32 `json:"u_id"` 	// 用户ID: 只是表示买家客户端是 userId
	CustomId  int32 `json:"c_id"`
	ShopId    int32 `json:"s_id"`
	Username  string `json:"u_name"`
	Platform  string `json:"platform"`
	Type      string `json:"type,omitempty"` // 客户端类型：shop-门店端, customer-客户端
	StaffId   int32 `json:"staff_id"` // 只是表示门店客户端的员工Id
}

// 全局互斥锁，用于保护 rooms 字典的并发访问，确保多个 goroutine 同时访问 rooms 时的线程安全
var mu sync.Mutex

// newRoom 创建并返回一个新的房间实例
// 初始化所有必要的 channel 和 map
func newRoom(name string) *Room {
	return &Room{
		clientConns: make(map[*Client]bool),
		join:        make(chan *Client),
		leave:       make(chan *Client),
		forward:     make(chan *message), // 现在传输 message 结构体
		name:        name,                // 设置房间名称
	}
}

// GetRoomAndRun 获取指定名称的房间，如果不存在则创建新房间并启动
// 这是线程安全的函数，使用互斥锁保护全局房间字典
func GetRoomAndRun(name string) *Room {
	mu.Lock()
	defer mu.Unlock()

	if r, ok := rooms[name]; ok {
		return r
	}
	r := newRoom(name) // 传入房间名称
	rooms[name] = r

	// 启动房间的消息处理循环（在新的 goroutine 中）
	go r.run()
	return r
}

// run 是房间的核心消息处理循环
// 在独立的 goroutine 中运行，使用 select 语句处理三种类型的事件
func (r *Room) run() {
	for {
		select {
		// 处理客户端加入事件
		case client := <-r.join:
			// 检查channel是否已关闭（收到nil值）
			if client == nil {
				return // channel已关闭，退出goroutine
			}
			// 将新客户端添加到房间的客户端集合中
			r.clientConns[client] = true // 使用 true 作为值，实际上只使用 key（client 指针）
			// log.Printf("客户端 %v 加入房间 %v", client.name, r.name)
			log.Info().Str("clientName", client.name).Str("roomName", r.name).Msg("客户端加入房间")
			// 清空未读消息数
			go ClearUnreadCount(client)

		// 处理客户端离开事件
		case client := <-r.leave:
			// 检查channel是否已关闭（收到nil值）
			if client == nil {
				return // channel已关闭，退出goroutine
			}
			// 从房间的客户端集合中删除客户端
			delete(r.clientConns, client)
			// 关闭客户端的接收 channel，通知客户端的 write() 方法退出
			close(client.receive)
			log.Info().Str("clientName", client.name).Str("roomName", r.name).Msg("客户端离开房间")
			if len(r.clientConns) == 0 {
				log.Info().Str("roomName", r.name).Msg("房间没有客户端，关闭房间")
				close(r.forward)
				close(r.join)
				close(r.leave)
				delete(rooms, r.name)
				return // 退出run() goroutine，避免继续从已关闭的channel读取数据
			}

		// 处理消息转发事件
		case msg := <-r.forward:
			// 检查channel是否已关闭（收到nil值）
			if msg == nil {
				return // channel已关闭，退出goroutine
			}
			// 打印消息内容与发送者信息
			log.Debug().Str("sender", msg.sender.name).Str("message", string(msg.content)).Msg("Received message")

			// 直接存储原始消息内容到Redis，使用房间名作为键
			go StoreRawMessage(r.name, msg.content) // 异步存储，不阻塞消息转发
			// 创建最近联系人列表 或 更新最近联系人列表
			go UpdateRecentContacts(msg.sender)

			online := false
			// 将消息发送给房间中的所有客户端，但排除发送者自己
			for client := range r.clientConns {
				if client != msg.sender { // 高效的指针比较，排除发送者  // && client.userInfo.Type != msg.sender.userInfo.Type -- 排除同一类型的客户端
					// 将消息发送到每个客户端的接收 channel
					client.receive <- msg.content // 只发送消息内容，不包含发送者信息
					online = true
				}
			}

			// 本消息转发，目的是让客户与门店互相发消息，当有另一方不在线时，要累计未读消息数
			if !online {
				go IncrementUnreadCount(msg.sender, 1)
			}
		}
	}
}

// 查询当前房间的客户端数量
func (r *Room)clientCount() int {
	return len(r.clientConns)
}

// 统计所有房间里的客户端总数量
func GetTotalClientCount() int {
	totalCount := 0
	for _, room := range rooms {
		totalCount += len(room.clientConns)
	}
	return totalCount
}

// 用 Client.name 来查找房间
func GetRoomByClientName(clientName string) *Room {
	for _, room := range rooms {
		for client := range room.clientConns {
			if client.name == clientName {
				return room
			}
		}
	}
	return nil
}

// 用 Client.name 查询用户是否在线
func IsClientOnline(clientName string) bool {
	room := GetRoomByClientName(clientName)
	return room != nil
}


// ======================================== webSocket服务相关 ========================================
// WebSocket 相关常量定义
const (
	// WebSocket 连接的读写缓冲区大小（字节）
	socketBufferSize = 1024
	// 每个客户端消息接收 channel 的缓冲区大小
	messageBufferSize = 256
)

// WebSocket 升级器，用于将 HTTP 连接升级为 WebSocket 连接
// 设置读写缓冲区大小以优化性能
var upgrader = &websocket.Upgrader{
	ReadBufferSize:  socketBufferSize, // 读缓冲区大小
	WriteBufferSize: socketBufferSize, // 写缓冲区大小
	CheckOrigin: func(r *http.Request) bool {
		return true // 允许跨域连接
	},
}

// parseWebSocketProtocols 解析WebSocket子协议中的参数
// 客户端通过子协议传递多个Base64编码的JSON参数片段
func ParseWebSocketProtocols(protocols []string) (*RoomParams, error) {
	//log.Println("-------- ParseWebSocketProtocols ---------")
	// 查找包含房间参数的子协议
	for _, protocol := range protocols {
		//log.Printf("Parsing protocol: %s", protocol)
		if strings.HasPrefix(protocol, "chat,") {
			// 初始化参数结构体
			params := RoomParams{}

			// 分割协议字符串，获取各个参数部分
			parts := strings.Split(protocol, ",")
			if len(parts) < 2 {
				log.Warn().Str("protocol", protocol).Msg("协议格式错误，参数不足")
				continue
			}

			// 遍历所有参数部分（跳过第一个"chat"）
			for i := 1; i < len(parts); i++ {
				part := strings.TrimSpace(parts[i])
				if err := parseProtocolPart(part, &params); err != nil {
					// log.Printf("解析协议部分失败 %s: %v", part, err)
					log.Error().Err(err).Str("part", part).Msg("解析协议部分失败")
					continue
				}
			}

			// log.Printf("成功解析WebSocket协议参数: %+v\n", params)
			log.Info().Any("params", params).Msg("成功解析WebSocket协议参数")
			return &params, nil
		} else {
			log.Debug().Str("protocol", protocol).Msg("跳过非chat协议")
		}
	}

	return nil, fmt.Errorf("未找到有效的房间参数协议")
}

// parseProtocolPart 解析单个协议参数部分
func parseProtocolPart(part string, params *RoomParams) error {
	// 检查参数格式：p0-..., p1-..., p2-...
	if len(part) < 3 || !strings.Contains(part, "-") {
		return fmt.Errorf("参数格式错误: %s", part)
	}

	// 分割前缀和base64内容
	sepIndex := strings.Index(part, "-")
	if sepIndex == -1 {
		return fmt.Errorf("缺少分隔符: %s", part)
	}

	prefix := part[:sepIndex]
	encodedData := part[sepIndex+1:]

	// Base64解码 - 按优先级尝试不同的编码方式
	decodedBytes, err := base64.RawStdEncoding.DecodeString(encodedData)
	if err != nil {
		// 如果RawStdEncoding失败，尝试RawURLEncoding（URL安全的base64）
		decodedBytes, err = base64.RawURLEncoding.DecodeString(encodedData)
		if err != nil {
			// 如果RawURLEncoding也失败，尝试标准的StdEncoding（带填充）
			decodedBytes, err = base64.StdEncoding.DecodeString(encodedData)
			if err != nil {
				// 最后尝试URLEncoding
				decodedBytes, err = base64.URLEncoding.DecodeString(encodedData)
				if err != nil {
					return fmt.Errorf("Base64解码失败: %v", err)
				}
			}
		}
	}

	// URL解码
	decodedStr, err := url.QueryUnescape(string(decodedBytes))
	if err != nil {
		return fmt.Errorf("URL解码失败: %v", err)
	}

	// 根据前缀类型解析不同的参数
	switch prefix {
	case "p0":
		// 解析用户ID和自定义ID
		var p0Data struct {
			UserId   int32 `json:"u_id"`
			CustomId int32 `json:"c_id"`
		}
		if err := json.Unmarshal([]byte(decodedStr), &p0Data); err != nil {
			return fmt.Errorf("p0参数JSON解析失败: %v", err)
		}
		params.UserId = p0Data.UserId
		params.CustomId = p0Data.CustomId

	case "p1":
		// 解析店铺ID和用户名
		var p1Data struct {
			ShopId   int32  `json:"s_id"`
			Username string `json:"u_name"`
		}
		if err := json.Unmarshal([]byte(decodedStr), &p1Data); err != nil {
			return fmt.Errorf("p1参数JSON解析失败: %v", err)
		}
		params.ShopId = p1Data.ShopId
		params.Username = p1Data.Username

	case "p2":
		// 解析平台和类型
		var p2Data struct {
			Platform string `json:"platform"`
			Type     string `json:"type"`
		}
		if err := json.Unmarshal([]byte(decodedStr), &p2Data); err != nil {
			return fmt.Errorf("p2参数JSON解析失败: %v", err)
		}
		params.Platform = p2Data.Platform
		params.Type = p2Data.Type

	case "p3":
		// 解析平台和类型
		var p3Data struct {
			StaffId int32 `json:"staff_id"`
		}
		if err := json.Unmarshal([]byte(decodedStr), &p3Data); err != nil {
			return fmt.Errorf("p2参数JSON解析失败: %v", err)
		}
		params.StaffId = p3Data.StaffId

	// 当有 p4 参数时，解析 p4 参数

	default:
		log.Warn().Str("prefix", prefix).Msg("未知的参数前缀")
	}

	return nil
}


func (r *Room) HttpServe(w http.ResponseWriter, req *http.Request,roomParams *RoomParams) {
	// 将 HTTP 连接升级为 WebSocket 连接，指定接受的子协议
	responseHeader := http.Header{}
	// 这行代码的作用是在 WebSocket握手响应中添加 Sec-WebSocket-Protocol头，值为"chat"。这表示服务器接受并选择了"chat"子协议作为通信协议。当客户端在请求中提供多个子协议选项时，服务器需要在响应中指定它选择使用哪一个。这行代码确认服务器将使用基础的"chat"协议与客户端通信，这样客户端就知道应该使用哪种协议格式来解释后续的WebSocket消息。
	responseHeader.Add("Sec-WebSocket-Protocol", "chat") // 响应基础的 chat 协议
	
	socket, err := upgrader.Upgrade(w, req, responseHeader)
	if err != nil {
		log.Error().Err(err).Msg("Upgrade error")
		return
	}

	// -----------------------------------    生成客户端对象     ----------------------------------
	// 创建新的客户端对象，使用解析出的参数
	client := &Client{
		name:     clientName(roomParams),
		socket:   socket,                              // WebSocket 连接
		receive:  make(chan []byte, messageBufferSize), // 接收消息的缓冲 channel
		room:     r,                                   // 使用当前房间实例
		userInfo: UserInfo{
			UserId:   roomParams.UserId,
			Username: roomParams.Username,
			CustomId: roomParams.CustomId,
			ShopId:   roomParams.ShopId,
			Platform: roomParams.Platform,
			Type: roomParams.Type,
			StaffId: roomParams.StaffId,
		},
	}

	// 如果用户名为空，生成一个默认用户名
	if client.name == "" {
		client.name = fmt.Sprintf("user_%d", rand.Intn(9000000))
	}

	// 将客户端加入当前房间
	r.join <- client
	defer func() { r.leave <- client }()

	// 启动客户端的写消息 goroutine
	go client.write()
	// 在当前 goroutine 中处理客户端的读消息（阻塞直到连接断开）
	client.read()
}