// Redis模块 - 负责Redis连接管理和消息存储 package wsClient import ( "context" "encoding/json" "fmt" "os" "strconv" "time" "github.com/redis/go-redis/v9" "github.com/rs/zerolog/log" ) // Redis客户端实例,全局变量 var redisClient *redis.Client // Redis上下文 var ctx = context.Background() // 分数计算常量 const ( UnreadMultiplier = 10000.0 // 未读消息数乘数,用于小数部分存储(增大乘数以提高精度) ContactTTL = 7 * 24 * time.Hour // 最近联系人过期时间(7天) // 门店用户类型 UserTypeShop = "shop" // 顾客用户类型 UserTypeCustomer = "customer" ) // ContactInfo 表示联系人信息 type ContactInfo struct { ContactID int32 `json:"contact_id"` // 联系人ID Timestamp time.Time `json:"timestamp"` // 最后聊天时间 UnreadCount int `json:"unread_count"` // 未读消息数 ContactType string `json:"contact_type"` // 联系人类型(shop/customer) } // 从环境变量初始化Redis连接 func InitializeRedisFromEnv() error { // 从环境变量获取Redis配置,如果没有设置则使用默认值 redisAddr := os.Getenv("REDIS_ADDR") if redisAddr == "" { redisAddr = "localhost:6379" // 默认Redis地址 } redisPassword := os.Getenv("REDIS_PASSWORD") // 默认为空 redisDB := 0 // 默认数据库 if dbStr := os.Getenv("REDIS_DB"); dbStr != "" { if db, err := strconv.Atoi(dbStr); err == nil { redisDB = db } } return InitRedis(redisAddr, redisPassword, redisDB) } // ChatMessage 表示存储在Redis中的聊天消息结构 type ChatMessage struct { Content string `json:"content"` // 原始消息内容(JSON字符串) Timestamp time.Time `json:"timestamp"` // 存储时间戳 RoomName string `json:"room_name"` // 房间名称 } // InitRedis 初始化Redis连接 func InitRedis(addr, password string, db int) error { redisClient = redis.NewClient(&redis.Options{ Addr: addr, // Redis服务器地址 (例如: "localhost:6379") Password: password, // Redis密码 (如果没有密码则为空字符串) DB: db, // Redis数据库编号 (0-15) }) // 测试Redis连接 pong, err := redisClient.Ping(ctx).Result() if err != nil { log.Error().Err(err).Msg("Redis连接失败") return err } log.Info().Str("pong", pong).Msg("Redis连接成功") return nil } // StoreRawMessage 将原始消息内容存储到Redis中 // 使用房间名作为Redis列表的键,存储消息历史记录 func StoreRawMessage(roomName string, rawMessage []byte) error { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return nil } // 创建消息对象,包含原始消息内容和时间戳 chatMsg := ChatMessage{ Content: string(rawMessage), // 直接存储原始JSON字符串 Timestamp: time.Now(), RoomName: roomName, } // 将消息序列化为JSON msgBytes, err := json.Marshal(chatMsg) if err != nil { log.Error().Err(err).Msg("消息序列化失败") return err } // 使用房间名作为Redis键,将消息添加到列表的右端 // 键格式: "room_messages:{房间名}" key := "room_messages:" + roomName err = redisClient.RPush(ctx, key, msgBytes).Err() if err != nil { log.Error().Err(err).Str("key", key).Msg("消息存储到Redis失败") return err } // 设置过期时间(可选)- 7天后自动删除 redisClient.Expire(ctx, key, 7*24*time.Hour) log.Debug().Str("roomName", roomName).Msg("消息已存储到Redis") return nil } // StoreMessage 将消息存储到Redis中(保留旧接口以兼容) // 使用房间名作为Redis列表的键,存储消息历史记录 func StoreMessage(roomName, username, message string) error { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return nil } // 构建与新格式兼容的消息结构 msgContent := map[string]interface{}{ "name": username, "message": message, } msgBytes, _ := json.Marshal(msgContent) return StoreRawMessage(roomName, msgBytes) } // UpdateRecentContacts 更新最近联系人列表 // 使用浮点数分数同时存储时间戳和未读消息数 func UpdateRecentContacts(client *Client) error { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return nil } currentTime := time.Now().Unix() if client.userInfo.Type == UserTypeShop { // 更新门店的联系人列表(客户) key := getContactKey(UserTypeShop, client.userInfo.ShopId) // shop_chats:{ShopId} if key != "" { // 先获取当前未读消息数 // currentScore, err := redisClient.ZScore(ctx, key, strconv.Itoa(int(client.userInfo.CustomId))).Result() unreadCount := 0 // ---->>>>> 门店是主动方,所以未读消息数为0 // if err == nil { // // 如果成员已存在,保留当前的未读消息数 // _, unreadCount = parseScore(currentScore) // } score := calculateScore(currentTime, unreadCount) redisClient.ZAdd(ctx, key, redis.Z{ Score: score, Member: strconv.Itoa(int(client.userInfo.CustomId)), }) // 设置过期时间 redisClient.Expire(ctx, key, ContactTTL) } // 更新客户的联系人列表(门店) key = getContactKey(UserTypeCustomer, client.userInfo.UserId) // customer_chats:{UserId} if key != "" { // 先获取当前未读消息数 currentScore, err := redisClient.ZScore(ctx, key, strconv.Itoa(int(client.userInfo.ShopId))).Result() unreadCount := 0 if err == nil { // 如果成员已存在,保留当前的未读消息数 _, unreadCount = parseScore(currentScore) } score := calculateScore(currentTime, unreadCount) redisClient.ZAdd(ctx, key, redis.Z{ Score: score, Member: strconv.Itoa(int(client.userInfo.ShopId)), }) // 设置过期时间 redisClient.Expire(ctx, key, ContactTTL) } } if client.userInfo.Type == UserTypeCustomer { // 更新客户的联系人列表(门店) key := getContactKey(UserTypeCustomer, client.userInfo.UserId) // customer_chats:{UserId} if key != "" { // 先获取当前未读消息数 // currentScore, err := redisClient.ZScore(ctx, key, strconv.Itoa(int(client.userInfo.ShopId))).Result() unreadCount := 0 // ---->>>>> 客户是主动方,所以未读消息数为0 // if err == nil { // // 如果成员已存在,保留当前的未读消息数 // _, unreadCount = parseScore(currentScore) // } score := calculateScore(currentTime, unreadCount) redisClient.ZAdd(ctx, key, redis.Z{ Score: score, Member: strconv.Itoa(int(client.userInfo.ShopId)), }) // 设置过期时间 redisClient.Expire(ctx, key, ContactTTL) } // 更新门店的联系人列表(客户) key = getContactKey(UserTypeShop, client.userInfo.ShopId) // shop_chats:{ShopId} if key != "" { // 先获取当前未读消息数 currentScore, err := redisClient.ZScore(ctx, key, strconv.Itoa(int(client.userInfo.CustomId))).Result() unreadCount := 0 if err == nil { // 如果成员已存在,保留当前的未读消息数 _, unreadCount = parseScore(currentScore) } score := calculateScore(currentTime, unreadCount) redisClient.ZAdd(ctx, key, redis.Z{ Score: score, Member: strconv.Itoa(int(client.userInfo.CustomId)), }) // 设置过期时间 redisClient.Expire(ctx, key, ContactTTL) } } log.Debug().Str("clientName", client.name).Msg("最近联系人列表已更新") return nil } // 增加未读消息数 // 为接收者(对方)的联系人列表增加未读消息数 func IncrementUnreadCount(sender *Client, count int) error { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return nil } // 根据发送者的类型确定接收者的类型和ID var receiverType string var receiverID int32 var senderMember string if sender.userInfo.Type == UserTypeShop { // 门店发送消息给客户,接收者是客户 receiverType = UserTypeCustomer receiverID = sender.userInfo.UserId senderMember = strconv.Itoa(int(sender.userInfo.ShopId)) } else if sender.userInfo.Type == UserTypeCustomer { // 客户发送消息给门店,接收者是门店 receiverType = UserTypeShop receiverID = sender.userInfo.ShopId senderMember = strconv.Itoa(int(sender.userInfo.CustomId)) } else { log.Warn().Str("clientType", sender.userInfo.Type).Msg("未知的客户端类型") return nil } // 获取接收者的联系人列表键 key := getContactKey(receiverType, receiverID) if key == "" { log.Error().Str("receiverType", receiverType).Int32("receiverID", receiverID).Msg("无法生成联系人键") return nil } // 获取当前分数 currentScore, err := redisClient.ZScore(ctx, key, senderMember).Result() if err != nil { // 如果成员不存在,使用当前时间戳和0未读消息数创建 currentScore = calculateScore(time.Now().Unix(), 0) } // 解析当前的分数 _, currentUnread := parseScore(currentScore) // 增加未读消息数 newUnread := currentUnread + count // 重新计算分数 timestamp := time.Now().Unix() newScore := calculateScore(timestamp, newUnread) // 更新分数 redisClient.ZAdd(ctx, key, redis.Z{ Score: newScore, Member: senderMember, }) // 设置过期时间 redisClient.Expire(ctx, key, ContactTTL) // log.Printf("未读消息数已增加 - 接收者类型: %s, 接收者ID: %d, 发送者ID: %s, 未读数: %d, 新分数: %f", receiverType, receiverID, senderMember, newUnread, newScore) log.Debug(). Str("receiverType", receiverType). Int32("receiverID", receiverID). Str("senderMember", senderMember). Int("newUnread", newUnread). Float64("newScore", newScore). Msg("未读消息数已增加") return nil } // GetRoomMessages 获取房间的历史消息 // 返回指定数量的最新消息 func GetRoomMessages(roomName string, count int64) ([]ChatMessage, error) { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return nil, nil } key := "room_messages:" + roomName // 获取列表中的最新消息(从右端开始,即最新的消息) // -count 表示从列表末尾开始的count个元素 result, err := redisClient.LRange(ctx, key, -count, -1).Result() if err != nil { //log.Printf("从Redis获取消息失败: %v", err) log.Error().Err(err).Str("key", key).Msg("从Redis获取消息失败") return nil, err } var messages []ChatMessage for _, msgStr := range result { var msg ChatMessage if err := json.Unmarshal([]byte(msgStr), &msg); err != nil { // log.Printf("消息反序列化失败: %v", err) log.Error().Err(err).Str("msg", msgStr).Msg("消息反序列化失败") continue } messages = append(messages, msg) } return messages, nil } // GetRoomMessageCount 获取房间的消息总数 func GetRoomMessageCount(roomName string) (int64, error) { if redisClient == nil { return 0, nil } key := "room_messages:" + roomName return redisClient.LLen(ctx, key).Result() } // CloseRedis 关闭Redis连接 func CloseRedis() error { if redisClient != nil { return redisClient.Close() } return nil } // calculateScore 计算有序集合的分数,同时存储时间戳和未读消息数 // 整数部分:Unix时间戳,小数部分:未读消息数(除以乘数) func calculateScore(timestamp int64, unreadCount int) float64 { return float64(timestamp) + (float64(unreadCount) / UnreadMultiplier) } // parseScore 从分数中解析出时间戳和未读消息数 func parseScore(score float64) (timestamp int64, unreadCount int) { timestamp = int64(score) // 整数部分是时间戳 decimalPart := score - float64(timestamp) // 使用四舍五入来处理浮点数精度问题 unreadCount = int(decimalPart*UnreadMultiplier + 0.5) return } // 生成发送者和接收者的信息 // @param sender 发送者 // @param count 未读消息数 // @return 接收者类型, 接收者ID, 未读消息数 func GeneHaveSendInfo(sender *Client, count int) (string, int32, int32, int) { if sender.userInfo.Type == UserTypeShop { // 门店发送消息给 客户 return UserTypeCustomer, sender.userInfo.ShopId, sender.userInfo.UserId, count } else if sender.userInfo.Type == UserTypeCustomer { // 客户发送消息给 门店 return UserTypeShop, sender.userInfo.ShopId, sender.userInfo.UserId, count } return "", 0, 0, 0 } /** * 生成联系人的Redis键 * @param userType 用户类型 * @param id 用户ID -- shopId 或 userId * @return 联系人的Redis键 */ func getContactKey(userType string, id int32) string { if userType == UserTypeShop { return "shop_chats:" + strconv.Itoa(int(id)) } else if userType == UserTypeCustomer { return "customer_chats:" + strconv.Itoa(int(id)) } return "" } // GetRecentContacts 获取用户的最近联系人列表 // 返回按最后聊天时间排序的联系人列表(最新的在前) func GetRecentContacts(userType string, userID int32, limit int64) ([]ContactInfo, error) { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return nil, nil } key := getContactKey(userType, userID) if key == "" { return nil, fmt.Errorf("无效的用户类型: %s", userType) } // 获取最近的联系人(按分数降序,最新的在前) result, err := redisClient.ZRevRangeWithScores(ctx, key, 0, limit-1).Result() if err != nil { log.Error().Err(err).Str("key", key).Msg("获取最近联系人失败") return nil, err } var contacts []ContactInfo for _, item := range result { contactID, err := strconv.Atoi(item.Member.(string)) if err != nil { // log.Printf("解析联系人ID失败: %v", err) log.Error().Err(err).Any("member", item.Member).Msg("解析联系人ID失败") continue } timestamp, unreadCount := parseScore(item.Score) // 确定联系人的类型 contactType := UserTypeCustomer if userType == UserTypeCustomer { contactType = UserTypeShop } contact := ContactInfo{ ContactID: int32(contactID), Timestamp: time.Unix(timestamp, 0), UnreadCount: unreadCount, ContactType: contactType, } contacts = append(contacts, contact) } return contacts, nil } // GetTotalUnreadCount 获取用户的总未读消息数 func GetTotalUnreadCount(userType string, userID int32) (int, error) { if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return 0, nil } key := getContactKey(userType, userID) if key == "" { return 0, fmt.Errorf("无效的用户类型: %s", userType) } // 获取所有联系人的分数 result, err := redisClient.ZRangeWithScores(ctx, key, 0, -1).Result() if err != nil { log.Error().Err(err).Str("key", key).Msg("获取联系人分数失败") return 0, err } totalUnread := 0 for _, item := range result { _, unreadCount := parseScore(item.Score) totalUnread += unreadCount } return totalUnread, nil } /** * 清除指定联系人的未读消息数 * @param client 客户端 * @return 错误 */ func ClearUnreadCount(client *Client) error { id := client.userInfo.ShopId contactID := client.userInfo.CustomId if client.userInfo.Type == UserTypeCustomer { id = client.userInfo.UserId contactID = client.userInfo.ShopId } if redisClient == nil { log.Warn().Msg("Redis客户端未初始化") return fmt.Errorf("redis client is not initialized") } key := getContactKey(client.userInfo.Type, id) if key == "" { log.Error().Str("userType", client.userInfo.Type).Msg("无效的用户类型") return fmt.Errorf("invalid user type: %s", client.userInfo.Type) } contactIDStr := strconv.Itoa(int(contactID)) // 获取当前分数 currentScore, err := redisClient.ZScore(ctx, key, contactIDStr).Result() if err != nil { if err == redis.Nil { // 如果联系人不存在,这并非一个错误,只是无需清理。 // log.Printf("Contact not found, no unread count to clear - UserType: %s, UserID: %d, ContactID: %d", client.userInfo.Type, id, contactID) log.Debug(). Str("userType", client.userInfo.Type). Int32("userID", id). Int32("contactID", contactID). Msg("Contact not found, no unread count to clear") return nil } // 对于其他类型的错误(如连接问题),则返回错误。 // log.Printf("failed to get score for contact - UserType: %s, UserID: %d, ContactID: %d, Error: %v",client.userInfo.Type, id, contactID, err) log.Error(). Err(err). Str("userType", client.userInfo.Type). Int32("userID", id). Int32("contactID", contactID). Msg("failed to get score for contact") return err } // 解析当前分数,保留时间戳,清除未读消息数 timestamp, _ := parseScore(currentScore) newScore := calculateScore(timestamp, 0) // 更新分数 if err := redisClient.ZAdd(ctx, key, redis.Z{ Score: newScore, Member: contactIDStr, }).Err(); err != nil { log.Error(). Err(err). Str("userType", client.userInfo.Type). Int32("userID", id). Int32("contactID", contactID). Msg("failed to clear unread count") return err } // log.Printf("未读消息数已清除 - 用户类型: %s, 用户ID: %d, 联系人ID: %d", client.userInfo.Type, id, contactID) log.Info(). Str("userType", client.userInfo.Type). Int32("userID", id). Int32("contactID", contactID). Msg("未读消息数已清除") return nil }