From b2c02c1db32f507c84dc996451bee7218ce46de3 Mon Sep 17 00:00:00 2001 From: lq Date: Wed, 16 Jul 2025 10:33:59 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E9=80=9A=E8=AF=9D=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E7=9A=84=E7=8A=B6=E6=80=81=E5=AD=97=E6=AE=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env | 8 +++ controller/websocket_controller.go | 105 ++++++++++++++++++++++++++++- go.mod | 6 ++ go.sum | 12 ++++ main.go | 25 ++++++- models/messages.go | 16 +++++ route/route.go | 16 +++-- 7 files changed, 181 insertions(+), 7 deletions(-) diff --git a/.env b/.env index 742e66e..5753c5f 100644 --- a/.env +++ b/.env @@ -2,6 +2,14 @@ REDIS_ADDR=127.0.0.1:6379 REDIS_PASSWORD= +# [Mysql] +DB_DSN=root:root@tcp(localhost:3310)/z_xk?charset=utf8mb4&parseTime=True&loc=Local +DB_HOST=127.0.0.1 +DB_PORT=3306 +DB_USER=root +DB_PASSWORD=root +DB_NAME=z_xk + # [App] PORT=12080 diff --git a/controller/websocket_controller.go b/controller/websocket_controller.go index a64612e..997c4f9 100644 --- a/controller/websocket_controller.go +++ b/controller/websocket_controller.go @@ -7,12 +7,14 @@ import ( "log" "net/http" "os" + "strconv" "sync" "time" "github.com/gin-gonic/gin" "github.com/go-redis/redis/v8" "github.com/gorilla/websocket" + "gorm.io/gorm" "xk-websocket/models" "xk-websocket/utils" ) @@ -35,12 +37,14 @@ type WebSocketController struct { Port string RedisCtx context.Context Logger *log.Logger + DB *gorm.DB // 数据库连接 } -func NewWebSocketController() *WebSocketController { +func NewWebSocketController(db *gorm.DB) *WebSocketController { return &WebSocketController{ Clients: make(map[string]*websocket.Conn), RedisCtx: context.Background(), + DB: db, // 初始化数据库连接 } } @@ -711,6 +715,7 @@ func (c *WebSocketController) BindHandler(ctx *gin.Context) { req.UserID, req.ClientID, time.Since(start)) } +// SendToUserHandler 通过用户ID发送消息处理器（支持通话信令） // SendToUserHandler 通过用户ID发送消息处理器（支持通话信令） func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) { start := time.Now() @@ -732,6 +737,24 @@ func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) { 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, + MessageType: payload.MessageType, + MessageContent: payload.MessageContent, + CallID: payload.CallID, + CallStatus: payload.CallStatus, + } + + // 保存到数据库 + if err := c.DB.Create(&chatMsg).Error; err != nil { + log.Printf("❌ 保存消息到数据库失败: %v", err) + } else { + log.Printf("💾 消息保存到数据库成功: ID=%d", chatMsg.ID) + } + // 创建基础消息结构 clientMsg := models.ClientReceivedMessage{ SenderID: "system", @@ -767,6 +790,86 @@ func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) { log.Printf("✅ API用户发送请求完成: Duration=%s", time.Since(start)) } +// 获取聊天记录（分页） +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), + }) +} + +// 同步聊天记录（全部） +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, + }) +} + // 添加客户端连接 func (c *WebSocketController) addClient(clientID string, conn *websocket.Conn) { c.ClientsMux.Lock() diff --git a/go.mod b/go.mod index 33e845d..45b9017 100644 --- a/go.mod +++ b/go.mod @@ -8,9 +8,12 @@ require ( github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 github.com/joho/godotenv v1.5.1 + gorm.io/driver/mysql v1.6.0 + gorm.io/gorm v1.30.0 ) require ( + filippo.io/edwards25519 v1.1.0 // indirect github.com/bytedance/sonic v1.11.6 // indirect github.com/bytedance/sonic/loader v0.1.1 // indirect github.com/cespare/xxhash/v2 v2.1.2 // indirect @@ -22,7 +25,10 @@ require ( github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.20.0 // indirect + github.com/go-sql-driver/mysql v1.8.1 // indirect github.com/goccy/go-json v0.10.2 // indirect + github.com/jinzhu/inflection v1.0.0 // indirect + github.com/jinzhu/now v1.1.5 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/leodido/go-urn v1.4.0 // indirect diff --git a/go.sum b/go.sum index 467bc36..2b2b095 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc0= github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4= github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM= @@ -31,6 +33,8 @@ github.com/go-playground/validator/v10 v10.20.0 h1:K9ISHbSaI0lyB2eWMPJo+kOS/FBEx github.com/go-playground/validator/v10 v10.20.0/go.mod h1:dbuPbCMFw/DrkbEynArYaCwl3amGuJotoKCe95atGMM= github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI= github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo= +github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= +github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg= github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU= @@ -40,6 +44,10 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= +github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= +github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= +github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= @@ -109,5 +117,9 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gorm.io/driver/mysql v1.6.0 h1:eNbLmNTpPpTOVZi8MMxCi2aaIm0ZpInbORNXDwyLGvg= +gorm.io/driver/mysql v1.6.0/go.mod h1:D/oCC2GWK3M/dqoLxnOlaNKmXz8WNTfcS9y5ovaSqKo= +gorm.io/gorm v1.30.0 h1:qbT5aPv1UH8gI99OsRlvDToLxW5zR7FzS9acZDOZcgs= +gorm.io/gorm v1.30.0/go.mod h1:8Z33v652h4//uMA76KjeDH8mJXPm1QNCYrMeatR0DOE= nullprogram.com/x/optparse v1.0.0/go.mod h1:KdyPE+Igbe0jQUrVfMqDMeJQIJZEuyV7pjYmp6pbG50= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/main.go b/main.go index 0131a19..23da2d4 100644 --- a/main.go +++ b/main.go @@ -2,11 +2,12 @@ package main import ( "flag" + "gorm.io/driver/mysql" + "gorm.io/gorm" "log" "os" "runtime/debug" "time" - "xk-websocket/controller" "xk-websocket/route" "xk-websocket/utils" @@ -48,6 +49,23 @@ func setupLogger() { log.SetOutput(fileWriter) } +// 初始化数据库连接 +func initDB() *gorm.DB { + dsn := utils.GetEnv("DB_DSN", "root:password@tcp(localhost:3306)/xk_chat?charset=utf8mb4&parseTime=True&loc=Local") + db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{}) + if err != nil { + log.Fatalf("❌ 数据库连接失败: %v", err) + } + + //// 自动迁移表结构 + //if err := db.AutoMigrate(&models.XkChatMessage{}); err != nil { + // log.Fatalf("❌ 数据库迁移失败: %v", err) + //} + + log.Println("✅ 数据库连接成功") + return db +} + func main() { // 初始化日志系统（必须放在最先） setupLogger() @@ -68,8 +86,11 @@ func main() { flag.StringVar(&nodeID, "nodeId", "", "节点标识符") flag.Parse() + // 初始化数据库 + db := initDB() + // 初始化控制器 - wsCtrl := controller.NewWebSocketController() + wsCtrl := controller.NewWebSocketController(db) // 设置参数优先级 if nodeID != "" { diff --git a/models/messages.go b/models/messages.go index 45acc1d..4cff003 100644 --- a/models/messages.go +++ b/models/messages.go @@ -1,5 +1,7 @@ package models +import "time" + // 统一消息结构（发送方使用） type SendMessagePayload struct { RequestType string `json:"request_type"` // 消息类型 @@ -15,6 +17,7 @@ type SendMessagePayload struct { // 通过用户ID发送消息的结构 type SendToUserPayload struct { + RoomId string `json:"room_id"` // 发送者用户ID SenderUserID string `json:"sender_user_id"` // 发送者用户ID ReceiverUserID string `json:"receiver_user_id"` // 接收者用户ID MessageType int `json:"message_type"` // 消息类型（0-8） @@ -24,6 +27,19 @@ type SendToUserPayload struct { CallStatus string `json:"call_status,omitempty"` // 通话状态:invite/accepted/rejected/ended/candidate/offer/answer } +// 聊天消息记录模型 +type XkChatMessage struct { + ID uint `gorm:"primaryKey" json:"id"` + RoomId string `gorm:"type:varchar(100);index" json:"room_id"` // 房间ID + SenderUserID string `gorm:"type:varchar(100);index" json:"sender_user_id"` // 发送者用户ID + ReceiverUserID string `gorm:"type:varchar(100);index" json:"receiver_user_id"` // 接收者用户ID + MessageType int `gorm:"type:int" json:"message_type"` // 消息类型（0-8） + MessageContent string `gorm:"type:text" json:"message_content"` // 消息内容 + CallID string `gorm:"type:varchar(100);index" json:"call_id"` // 通话唯一ID + CallStatus string `gorm:"type:varchar(50)" json:"call_status"` // 通话状态 + CreatedAt time.Time `gorm:"autoCreateTime" json:"created_at"` // 创建时间 +} + // 客户端接收消息结构（最终格式） type ClientReceivedMessage struct { SenderID string `json:"sender_id"` // 发送者连接ID diff --git a/route/route.go b/route/route.go index e47295f..62e1d82 100644 --- a/route/route.go +++ b/route/route.go @@ -10,8 +10,16 @@ func SetupRoutes(router *gin.Engine, wsCtrl *controller.WebSocketController) { router.Use(gin.Recovery()) router.GET("/ws", wsCtrl.HandleWebSocket) - router.POST("/api/send", wsCtrl.SendMessageHandler) - router.POST("/api/bind", wsCtrl.BindHandler) - router.POST("/api/send-to-user", wsCtrl.SendToUserHandler) - router.GET("/api/health", wsCtrl.HealthHandler) + // API 分组路由 + apiGroup := router.Group("/api") + { + apiGroup.POST("/send", wsCtrl.SendMessageHandler) + apiGroup.POST("/bind", wsCtrl.BindHandler) + apiGroup.POST("/send-to-user", wsCtrl.SendToUserHandler) + apiGroup.GET("/health", wsCtrl.HealthHandler) + + // 新增的消息路由 + apiGroup.GET("/messages", wsCtrl.GetMessagesHandler) + apiGroup.GET("/messages/sync", wsCtrl.SyncMessagesHandler) + } }