587 lines
16 KiB
Go
587 lines
16 KiB
Go
/**
|
||
* package service
|
||
* 作用：封装核心业务逻辑，包括数据库操作、Redis 消息分发、URL 抓取和多端同步。
|
||
*/
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"regexp"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"xk-websocket-v2/internal/manager"
|
||
"xk-websocket-v2/internal/model"
|
||
"xk-websocket-v2/internal/turnserver"
|
||
"xk-websocket-v2/internal/utils"
|
||
|
||
"github.com/go-redis/redis/v8"
|
||
"github.com/spf13/viper"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// ChatService 业务服务结构体
|
||
type ChatService struct {
|
||
DB *gorm.DB
|
||
Redis *redis.Client
|
||
}
|
||
|
||
// ChatSvc 全局单例
|
||
var ChatSvc *ChatService
|
||
|
||
/**
|
||
* InitChatService
|
||
* 功能：初始化聊天服务，并启动 Redis 集群消息订阅。
|
||
*/
|
||
func InitChatService(db *gorm.DB, rdb *redis.Client) {
|
||
ChatSvc = &ChatService{DB: db, Redis: rdb}
|
||
// 启动 Redis 订阅监听协程
|
||
go ChatSvc.SubscribeClusterMessages()
|
||
}
|
||
|
||
/**
|
||
* BindUser
|
||
* 功能：处理用户绑定逻辑，更新本地连接管理器和 Redis 路由表。
|
||
* @param client 本地客户端连接
|
||
* @param userID 用户ID
|
||
*/
|
||
func (s *ChatService) BindUser(client *manager.Client, userID string) {
|
||
// 1. 本地内存绑定
|
||
manager.Manager.BindUser(client.ID, userID)
|
||
|
||
// 2. 更新 Redis 路由表：记录 UserID 所在的 NodeID
|
||
// 用于 DispatchMessage 判断用户在哪个节点
|
||
nodeID := viper.GetString("app.node_id")
|
||
ctx := context.Background()
|
||
key := model.KeyUserNodeMap + userID
|
||
|
||
// 设置 24 小时过期
|
||
s.Redis.Set(ctx, key, nodeID, 24*time.Hour)
|
||
}
|
||
|
||
/**
|
||
* BindUserByClientID
|
||
* 功能：根据 ClientID 查找连接并绑定用户（用于 HTTP 绑定接口）。
|
||
*/
|
||
func (s *ChatService) BindUserByClientID(clientID, userID string) {
|
||
manager.Manager.BindUser(clientID, userID)
|
||
ctx := context.Background()
|
||
key := model.KeyUserNodeMap + userID
|
||
nodeID := viper.GetString("app.node_id")
|
||
s.Redis.Set(ctx, key, nodeID, 24*time.Hour)
|
||
}
|
||
|
||
/**
|
||
* IsUserOnline
|
||
* 功能：检查用户是否在线 (查询 Redis)。
|
||
*/
|
||
func (s *ChatService) IsUserOnline(userID string) bool {
|
||
ctx := context.Background()
|
||
key := model.KeyUserNodeMap + userID
|
||
exists, _ := s.Redis.Exists(ctx, key).Result()
|
||
return exists > 0
|
||
}
|
||
|
||
/**
|
||
* HandleUserMessage
|
||
* 功能：处理用户发来的消息（入口函数）。包含多端同步、URL 抓取、持久化和转发逻辑。
|
||
* @param senderClient 发送消息的客户端（用于识别来源设备）
|
||
* @param req 消息请求体
|
||
* @returns 持久化后的消息（信令/群通知/被拦截时返回 nil）
|
||
*/
|
||
func (s *ChatService) HandleUserMessage(senderClient *manager.Client, req *model.SendMessageReq) *model.ChatMessage {
|
||
// 1. [关键] 拦截通话信令：处理多端同步逻辑
|
||
if req.CallStatus == "accepted" {
|
||
s.NotifyOtherDevices(senderClient.UserID, senderClient.ID, req.CallID)
|
||
}
|
||
|
||
// [FIXED] 核心修复：直接使用前端传递的 Extra
|
||
var extraData string = req.Extra
|
||
|
||
// 2. URL 识别与抓取逻辑
|
||
if req.MessageType == 0 {
|
||
url := extractURL(req.Content)
|
||
if url != "" {
|
||
meta := s.scrapeURL(url)
|
||
if meta != nil {
|
||
metaJson, _ := json.Marshal(meta)
|
||
extraData = string(metaJson)
|
||
log.Printf("🌐 抓取成功: %s", meta.Title)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 3. 消息持久化 (MySQL)
|
||
// 信令消息(6) 和 群通知(1000+) 不持久化
|
||
if req.MessageType != model.MessageTypeSignal && req.MessageType != model.MessageTypeGroupNotif {
|
||
// ... (省略常规消息持久化逻辑，保持原样) ...
|
||
isGroupMessage := false
|
||
if req.RoomID != "" {
|
||
room, err := RoomSvc.GetRoom(req.RoomID)
|
||
if err == nil && room.RoomType == "group" {
|
||
isGroupMessage = true
|
||
|
||
// 验证群成员身份
|
||
memberIDs, mErr := RoomSvc.GetRoomMembers(req.RoomID)
|
||
if mErr != nil {
|
||
return nil
|
||
}
|
||
|
||
isMember := false
|
||
for _, memberID := range memberIDs {
|
||
if memberID == senderClient.UserID {
|
||
isMember = true
|
||
break
|
||
}
|
||
}
|
||
|
||
if !isMember {
|
||
// 发送错误消息
|
||
errorMsg := model.WsPayload{
|
||
RequestType: "error",
|
||
Data: map[string]interface{}{
|
||
"message": "您已被移出群聊，无法发送消息",
|
||
"code": "FORBIDDEN",
|
||
},
|
||
}
|
||
errorBytes, _ := json.Marshal(errorMsg)
|
||
s.DispatchMessage(senderClient.UserID, errorBytes)
|
||
return nil
|
||
}
|
||
}
|
||
}
|
||
|
||
receiverUserID := req.ReceiverUserID
|
||
if isGroupMessage {
|
||
receiverUserID = req.RoomID
|
||
}
|
||
|
||
// 私聊：接收方已拉黑发送方则拒绝
|
||
if !isGroupMessage && req.ReceiverUserID != "" && ContactSvc != nil {
|
||
if ContactSvc.IsBlocked(req.ReceiverUserID, senderClient.UserID) {
|
||
errorMsg := model.WsPayload{
|
||
RequestType: "error",
|
||
Data: map[string]interface{}{
|
||
"message": "消息已发出，但被对方拒收",
|
||
"code": "BLOCKED",
|
||
},
|
||
}
|
||
errorBytes, _ := json.Marshal(errorMsg)
|
||
s.DispatchMessage(senderClient.UserID, errorBytes)
|
||
return nil
|
||
}
|
||
}
|
||
|
||
msg := model.ChatMessage{
|
||
RoomID: req.RoomID,
|
||
SenderUserID: senderClient.UserID,
|
||
SenderIP: senderClient.RemoteIP,
|
||
IPLocation: utils.GetIPLocation(senderClient.RemoteIP),
|
||
ReceiverUserID: receiverUserID,
|
||
MessageType: req.MessageType,
|
||
Content: req.Content,
|
||
Extra: extraData,
|
||
Duration: req.Duration,
|
||
CallID: req.CallID,
|
||
CallStatus: req.CallStatus,
|
||
}
|
||
|
||
if err := s.DB.Create(&msg).Error; err != nil {
|
||
log.Printf("❌ 消息持久化失败: %v", err)
|
||
}
|
||
|
||
// 更新会话列表逻辑 (保持原样)
|
||
if ConversationSvc != nil && (req.MessageType == model.MessageTypeText ||
|
||
req.MessageType == model.MessageTypeImage ||
|
||
req.MessageType == model.MessageTypeAudio ||
|
||
req.MessageType == model.MessageTypeVideo ||
|
||
req.MessageType == model.MessageTypeFile) {
|
||
|
||
senderID := senderClient.UserID
|
||
if isGroupMessage {
|
||
memberIDs, mErr := RoomSvc.GetRoomMembers(req.RoomID)
|
||
if mErr == nil {
|
||
for _, memberID := range memberIDs {
|
||
if memberID != senderID {
|
||
_ = ConversationSvc.UpsertConversationOnMessage(memberID, req.RoomID, req.RoomID, msg, false)
|
||
}
|
||
}
|
||
_ = ConversationSvc.UpsertConversationOnMessage(senderID, req.RoomID, req.RoomID, msg, true)
|
||
}
|
||
} else {
|
||
receiverID := req.ReceiverUserID
|
||
if senderID != "" && receiverID != "" {
|
||
_ = ConversationSvc.UpsertConversationOnMessage(senderID, receiverID, req.RoomID, msg, true)
|
||
_ = ConversationSvc.UpsertConversationOnMessage(receiverID, senderID, req.RoomID, msg, false)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 构建推送消息
|
||
pushMsg := model.WsPayload{
|
||
RequestType: "receive_message",
|
||
Data: msg,
|
||
}
|
||
msgBytes, _ := json.Marshal(pushMsg)
|
||
|
||
// 分发消息 (保持原样)
|
||
if req.RoomID != "" {
|
||
room, err := RoomSvc.GetRoom(req.RoomID)
|
||
if err == nil && room.RoomType == "group" {
|
||
memberIDs, mErr := RoomSvc.GetRoomMembers(req.RoomID)
|
||
if mErr == nil {
|
||
senderID := senderClient.UserID
|
||
for _, uid := range memberIDs {
|
||
if uid != senderID {
|
||
s.DispatchMessage(uid, msgBytes)
|
||
}
|
||
}
|
||
|
||
// 处理@通知
|
||
s.handleMentionNotification(senderClient.UserID, req.RoomID, extraData, msg)
|
||
return &msg
|
||
}
|
||
}
|
||
}
|
||
s.DispatchMessage(req.ReceiverUserID, msgBytes)
|
||
return &msg
|
||
|
||
} else {
|
||
// ==========================================
|
||
// 信令消息处理 (MessageType = 6) [核心修复]
|
||
// ==========================================
|
||
|
||
sigMsg := model.ChatMessage{
|
||
RoomID: req.RoomID, // 补全 RoomID
|
||
SenderUserID: senderClient.UserID,
|
||
ReceiverUserID: req.ReceiverUserID,
|
||
MessageType: req.MessageType,
|
||
Content: req.Content,
|
||
Extra: extraData,
|
||
CallID: req.CallID,
|
||
CallStatus: req.CallStatus,
|
||
CreatedAt: time.Now(),
|
||
}
|
||
pushMsg := model.WsPayload{
|
||
RequestType: "receive_message",
|
||
Data: sigMsg,
|
||
}
|
||
msgBytes, _ := json.Marshal(pushMsg)
|
||
|
||
// 修复逻辑：如果是群信令 (req.RoomID 不为空 且 ReceiverUserID 为空)，则广播
|
||
// 场景：Group Call Invite, Sync State
|
||
if req.RoomID != "" && req.ReceiverUserID == "" {
|
||
// 广播给群成员
|
||
memberIDs, mErr := RoomSvc.GetRoomMembers(req.RoomID)
|
||
if mErr == nil {
|
||
senderID := senderClient.UserID
|
||
for _, uid := range memberIDs {
|
||
// 不发给自己
|
||
if uid != senderID {
|
||
s.DispatchMessage(uid, msgBytes)
|
||
}
|
||
}
|
||
log.Printf("📡 [WebRTC] 群广播信令: RoomID=%s Action=%s", req.RoomID, req.CallStatus)
|
||
}
|
||
} else {
|
||
// 定向信令 (P2P)
|
||
// 场景：Offer, Answer, Candidate
|
||
s.DispatchMessage(req.ReceiverUserID, msgBytes)
|
||
log.Printf("📡 [WebRTC] 定向信令: To=%s Action=%s", req.ReceiverUserID, req.CallStatus)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 辅助：正则提取第一个 URL
|
||
func extractURL(text string) string {
|
||
re := regexp.MustCompile(`https?://[^\s]+`)
|
||
return re.FindString(text)
|
||
}
|
||
|
||
// 辅助：优化后的网页抓取 (Title, Desc, Image)
|
||
// 解决乱码和图片路径问题
|
||
func (s *ChatService) scrapeURL(url string) *model.UrlMeta {
|
||
client := &http.Client{Timeout: 5 * time.Second}
|
||
resp, err := client.Get(url)
|
||
if err != nil {
|
||
return nil
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode != 200 {
|
||
return nil
|
||
}
|
||
|
||
// 读取更多内容 (100KB) 以确保 meta 标签完整
|
||
buf := make([]byte, 100*1024)
|
||
n, _ := io.ReadFull(resp.Body, buf)
|
||
|
||
// 简单的编码修复：如果不是有效UTF-8，直接忽略或返回空（生产环境应用 charset decoder）
|
||
html := string(buf[:n])
|
||
if !utf8.ValidString(html) {
|
||
// 尝试作为 GBK 处理（这里简化处理，直接返回 nil 或 url 本身）
|
||
return nil
|
||
}
|
||
|
||
meta := &model.UrlMeta{Url: url}
|
||
|
||
// 优化正则：支持单双引号，支持换行
|
||
titleRe := regexp.MustCompile(`(?i)<title>(.*?)</title>`)
|
||
matches := titleRe.FindStringSubmatch(html)
|
||
if len(matches) > 1 {
|
||
meta.Title = strings.TrimSpace(matches[1])
|
||
}
|
||
|
||
// 优化 og:image 正则
|
||
imgRe := regexp.MustCompile(`(?i)<meta\s+property=["']og:image["']\s+content=["'](.*?)["']`)
|
||
imgMatches := imgRe.FindStringSubmatch(html)
|
||
if len(imgMatches) > 1 {
|
||
meta.Image = imgMatches[1]
|
||
}
|
||
|
||
// 优化 description 正则
|
||
descRe := regexp.MustCompile(`(?i)<meta\s+name=["']description["']\s+content=["'](.*?)["']`)
|
||
descMatches := descRe.FindStringSubmatch(html)
|
||
if len(descMatches) > 1 {
|
||
meta.Description = strings.TrimSpace(descMatches[1])
|
||
}
|
||
|
||
if meta.Title == "" {
|
||
return nil
|
||
}
|
||
|
||
// 相对路径处理
|
||
if meta.Image != "" && !strings.HasPrefix(meta.Image, "http") {
|
||
// 简单拼接，不严谨但够用
|
||
if strings.HasPrefix(meta.Image, "/") {
|
||
// 提取 domain
|
||
domainRe := regexp.MustCompile(`(https?://[^/]+)`)
|
||
domain := domainRe.FindString(url)
|
||
meta.Image = domain + meta.Image
|
||
}
|
||
}
|
||
|
||
return meta
|
||
}
|
||
|
||
/**
|
||
* NotifyOtherDevices
|
||
* 功能：通知同一用户的其他设备“电话已被接听”，实现多端协同。
|
||
*/
|
||
func (s *ChatService) NotifyOtherDevices(userID, currentClientID, callID string) {
|
||
// 获取其他在线设备ID
|
||
otherClientIDs := manager.Manager.GetOtherClients(userID, currentClientID)
|
||
|
||
if len(otherClientIDs) == 0 {
|
||
return
|
||
}
|
||
|
||
log.Printf("📱 [多端同步] 用户 %s 在 %s 接听，通知其他 %d 个设备取消振铃", userID, currentClientID, len(otherClientIDs))
|
||
|
||
// 构建“已在其他设备接听”的信令
|
||
cancelMsg := model.WsPayload{
|
||
RequestType: "receive_message",
|
||
Data: model.ChatMessage{
|
||
MessageType: 6, // 视频通话类型
|
||
CallStatus: "answered_elsewhere", // 特殊状态码
|
||
CallID: callID,
|
||
Content: "通话已在其他设备接听",
|
||
},
|
||
}
|
||
bytes, _ := json.Marshal(cancelMsg)
|
||
|
||
// 推送给其他设备
|
||
for _, cid := range otherClientIDs {
|
||
manager.Manager.SendToClient(cid, bytes)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* BroadcastChatMessage
|
||
* 功能：将聊天消息推送给房间成员或私聊对方（含发送者其他设备）
|
||
*/
|
||
func (s *ChatService) BroadcastChatMessage(msg model.ChatMessage) {
|
||
pushMsg := model.WsPayload{
|
||
RequestType: "receive_message",
|
||
Data: msg,
|
||
}
|
||
msgBytes, _ := json.Marshal(pushMsg)
|
||
|
||
if msg.RoomID != "" {
|
||
room, err := RoomSvc.GetRoom(msg.RoomID)
|
||
if err == nil && room.RoomType == "group" {
|
||
memberIDs, mErr := RoomSvc.GetRoomMembers(msg.RoomID)
|
||
if mErr == nil {
|
||
for _, uid := range memberIDs {
|
||
if uid != msg.SenderUserID {
|
||
s.DispatchMessage(uid, msgBytes)
|
||
}
|
||
}
|
||
s.DispatchMessage(msg.SenderUserID, msgBytes)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
if msg.ReceiverUserID != "" {
|
||
s.DispatchMessage(msg.ReceiverUserID, msgBytes)
|
||
}
|
||
s.DispatchMessage(msg.SenderUserID, msgBytes)
|
||
}
|
||
|
||
/**
|
||
* DispatchMessage
|
||
* 功能：将消息路由到目标用户（无论其在哪个节点）。
|
||
* 逻辑：先查 Redis 看用户在哪个节点，如果是本机则直接发，否则通过 Pub/Sub 转发。
|
||
*/
|
||
func (s *ChatService) DispatchMessage(targetUserID string, msgBytes []byte) {
|
||
ctx := context.Background()
|
||
|
||
// 1. 查询目标用户所在的节点ID
|
||
nodeID, err := s.Redis.Get(ctx, model.KeyUserNodeMap+targetUserID).Result()
|
||
|
||
currentNode := viper.GetString("app.node_id")
|
||
|
||
// 用户不在线或 Redis 记录过期
|
||
if err == redis.Nil || nodeID == "" {
|
||
return
|
||
}
|
||
|
||
// 2. 判断是否在本机
|
||
if nodeID == currentNode {
|
||
// 在本机，直接投递
|
||
manager.Manager.SendToUser(targetUserID, msgBytes)
|
||
} else {
|
||
// 3. 不在本机，通过 Redis 广播给集群
|
||
clusterMsg := model.ClusterMessage{
|
||
SenderNodeID: currentNode,
|
||
TargetUserID: targetUserID,
|
||
Payload: msgBytes,
|
||
}
|
||
bytes, _ := json.Marshal(clusterMsg)
|
||
s.Redis.Publish(ctx, model.ChanClusterBroadcast, bytes)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* SubscribeClusterMessages
|
||
* 功能：监听集群广播频道，处理来自其他节点的消息转发请求。
|
||
*/
|
||
func (s *ChatService) SubscribeClusterMessages() {
|
||
pubsub := s.Redis.Subscribe(context.Background(), model.ChanClusterBroadcast)
|
||
defer pubsub.Close()
|
||
|
||
ch := pubsub.Channel()
|
||
currentNode := viper.GetString("app.node_id")
|
||
|
||
for msg := range ch {
|
||
var clusterMsg model.ClusterMessage
|
||
if err := json.Unmarshal([]byte(msg.Payload), &clusterMsg); err != nil {
|
||
continue
|
||
}
|
||
|
||
// 忽略来自自己的消息
|
||
if clusterMsg.SenderNodeID == currentNode {
|
||
continue
|
||
}
|
||
|
||
// 修正：Payload 在 JSON 传输后变成了 interface{}，需要重新序列化才能发给 WebSocket
|
||
payloadBytes, _ := json.Marshal(clusterMsg.Payload)
|
||
|
||
// 尝试发送给本地用户
|
||
manager.Manager.SendToUser(clusterMsg.TargetUserID, payloadBytes)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* handleMentionNotification
|
||
* 功能：处理群聊消息中的@通知
|
||
* @param senderID 发送者ID
|
||
* @param roomID 房间ID
|
||
* @param extraData 消息的extra字段（JSON格式）
|
||
* @param msg 消息对象
|
||
*/
|
||
func (s *ChatService) handleMentionNotification(senderID, roomID, extraData string, msg model.ChatMessage) {
|
||
if extraData == "" {
|
||
return
|
||
}
|
||
|
||
// 解析 extra 字段
|
||
var extra struct {
|
||
MentionUserIDs []string `json:"mention_user_ids"`
|
||
}
|
||
if err := json.Unmarshal([]byte(extraData), &extra); err != nil {
|
||
return
|
||
}
|
||
|
||
if len(extra.MentionUserIDs) == 0 {
|
||
return
|
||
}
|
||
|
||
// 获取发送者信息
|
||
var sender model.User
|
||
if err := s.DB.Where("id = ?", senderID).First(&sender).Error; err != nil {
|
||
return
|
||
}
|
||
sender.Password = ""
|
||
|
||
// 获取房间信息
|
||
var room model.ChatRoom
|
||
if err := s.DB.Where("room_id = ?", roomID).First(&room).Error; err != nil {
|
||
return
|
||
}
|
||
|
||
// 为每个被@的用户发送通知
|
||
for _, mentionUserID := range extra.MentionUserIDs {
|
||
// 不通知自己
|
||
if mentionUserID == senderID {
|
||
continue
|
||
}
|
||
|
||
// 构建@通知消息
|
||
notifMsg := model.WsPayload{
|
||
RequestType: "mention_notification",
|
||
Data: map[string]interface{}{
|
||
"type": "mention",
|
||
"room_id": roomID,
|
||
"room_name": room.RoomName,
|
||
"sender": sender,
|
||
"message": msg.Content,
|
||
"message_id": msg.ID,
|
||
"created_at": msg.CreatedAt,
|
||
},
|
||
}
|
||
notifBytes, _ := json.Marshal(notifMsg)
|
||
s.DispatchMessage(mentionUserID, notifBytes)
|
||
|
||
log.Printf("📢 [@通知] %s 在群 %s 中@了 %s", senderID, roomID, mentionUserID)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* GetICEServers
|
||
* 功能：获取 TURN 配置 (供 API 层调用)
|
||
*/
|
||
func (s *ChatService) GetICEServers(userID string) []model.ICEServerConfig {
|
||
if !viper.GetBool("turn.enabled") {
|
||
return []model.ICEServerConfig{}
|
||
}
|
||
ip := viper.GetString("turn.public_ip")
|
||
port := viper.GetInt("turn.listen_port")
|
||
username, credential := turnserver.GenerateCredentials(userID)
|
||
return []model.ICEServerConfig{
|
||
{
|
||
Urls: []string{fmt.Sprintf("turn:%s:%d", ip, port)},
|
||
Username: username,
|
||
Credential: credential,
|
||
},
|
||
}
|
||
}
|