message-server/wsClient/room.go

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

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

	"github.com/gorilla/websocket"
)

// 全局房间 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: 门店客户端是 staffId, 买家客户端是 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-客户端
}

// 全局互斥锁，用于保护 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
}

// GetRoom 获取指定名称的房间，如果不存在则创建新房间但不启动
// 注意：这个函数只创建房间但不启动 run() 方法，可能导致房间无法处理消息
// 建议使用 GetRoomAndRun() 替代
func GetRoom(roomName string) *Room {
	mu.Lock()
	defer mu.Unlock()

	if r, ok := rooms[roomName]; ok {
		return r
	}
	nr := newRoom(roomName) // 传入房间名称
	rooms[roomName] = nr
	// 注意：这里没有调用 go nr.run()，房间不会处理消息
	return nr
}

// run 是房间的核心消息处理循环
// 在独立的 goroutine 中运行，使用 select 语句处理三种类型的事件
func (r *Room) run() {
	for {
		select {
		// 处理客户端加入事件
		case client := <-r.join:
			// 将新客户端添加到房间的客户端集合中
			r.clientConns[client] = true // 使用 true 作为值，实际上只使用 key（client 指针）

		// 处理客户端离开事件
		case client := <-r.leave:
			// 从房间的客户端集合中删除客户端
			delete(r.clientConns, client)
			// 关闭客户端的接收 channel，通知客户端的 write() 方法退出
			close(client.receive)

		// 处理消息转发事件
		case msg := <-r.forward:
			// 打印消息内容与发送者信息
			log.Printf("Received message from %v: --- %s", msg.sender, string(msg.content))

			// 直接存储原始消息内容到Redis，使用房间名作为键
			go StoreRawMessage(r.name, msg.content) // 异步存储，不阻塞消息转发

			// 将消息发送给房间中的所有客户端，但排除发送者自己
			// 使用指针比较是最高效的方法，因为每个客户端都有唯一的内存地址
			for client := range r.clientConns {
				if client != msg.sender { // 高效的指针比较，排除发送者
					// 将消息发送到每个客户端的接收 channel
					client.receive <- msg.content // 只发送消息内容，不包含发送者信息
				}
			}
		}
	}
}

// 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.Printf("协议格式错误，参数不足: %s", protocol)
				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)
					continue
				}
			}

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

	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解码 - 使用RawStdEncoding来处理可能没有正确填充的base64字符串
	decodedBytes, err := base64.RawStdEncoding.DecodeString(encodedData)
	if err != nil {
		// 如果RawStdEncoding失败，尝试标准的StdEncoding（带填充）
		decodedBytes, err = base64.StdEncoding.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

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

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

	default:
		log.Printf("未知的参数前缀: %s", prefix)
	}

	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.Println("Upgrade error:", err)
		return
	}

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

	// 如果用户名为空，生成一个默认用户名
	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()
}

// ServeHTTP 处理 WebSocket 连接请求
// 实现了 http.Handler 接口，负责将 HTTP 连接升级为 WebSocket 连接，并创建新的客户端加入当前房间
func (r *Room) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	// 解析WebSocket子协议中的参数
	protocols := req.Header["Sec-Websocket-Protocol"]
	roomParams, err := ParseWebSocketProtocols(protocols)
	if err != nil {
		log.Printf("解析WebSocket协议参数失败: %v", err)
		http.Error(w, "Invalid WebSocket protocols", http.StatusBadRequest)
		return
	}

	// 验证房间名称是否匹配
	if roomParams.RoomName != r.name {
		log.Printf("房间名称不匹配: 期望 %s, 收到 %s", r.name, roomParams.RoomName)
		http.Error(w, "Room name mismatch", http.StatusBadRequest)
		return
	}

	// 将 HTTP 连接升级为 WebSocket 连接，指定接受的子协议
	responseHeader := http.Header{}
	responseHeader.Add("Sec-WebSocket-Protocol", "chat") // 响应基础聊天协议
	
	socket, err := upgrader.Upgrade(w, req, responseHeader)
	if err != nil {
		log.Println("Upgrade error:", err)
		return
	}

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

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

	log.Printf("新客户端连接: 用户名=%s, 用户ID=%d, 自定义ID=%d, 店铺ID=%d, 平台=%s, 客户端类型=%s\n\n",
		client.userInfo.Username, client.userInfo.UserId, client.userInfo.CustomId, client.userInfo.ShopId, client.userInfo.Platform, client.userInfo.Type)

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

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