commit 3eee37fa55814ec45f6ea0b0d7d3a1e7a7d0b813 Author: liqi Date: Tue Dec 2 21:00:26 2025 +0800 初始化功能 diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..35410ca --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,8 @@ +# 默认忽略的文件 +/shelf/ +/workspace.xml +# 基于编辑器的 HTTP 客户端请求 +/httpRequests/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..a462469 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/nl-im-service.iml b/.idea/nl-im-service.iml new file mode 100644 index 0000000..5e764c4 --- /dev/null +++ b/.idea/nl-im-service.iml @@ -0,0 +1,9 @@ + + + + + + + + + \ No newline at end of file diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 0000000..e2b1481 --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,128 @@ +package main + +import ( + "fmt" + "log" + "net/http" + "time" + + "xk-websocket-v2/internal/api" + "xk-websocket-v2/internal/manager" + "xk-websocket-v2/internal/model" + "xk-websocket-v2/internal/service" + "xk-websocket-v2/internal/turnserver" + "xk-websocket-v2/internal/ws" + + "github.com/gin-gonic/gin" + "github.com/go-redis/redis/v8" + "github.com/gorilla/websocket" + "github.com/spf13/viper" + "gorm.io/driver/mysql" + "gorm.io/gorm" +) + +// ... (initConfig, initDB, initRedis 保持不变) ... +func initConfig() { + viper.SetConfigName("config") + viper.SetConfigType("yaml") + viper.AddConfigPath("configs") + viper.AddConfigPath(".") + if err := viper.ReadInConfig(); err != nil { + log.Fatalf("❌ 无法读取配置文件: %v", err) + } +} + +func initDB() *gorm.DB { + dsn := viper.GetString("database.dsn") + db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{}) + if err != nil { + log.Fatalf("❌ 数据库连接失败: %v", err) + } + db.AutoMigrate(&model.ChatMessage{}) + return db +} + +func initRedis() *redis.Client { + rdb := redis.NewClient(&redis.Options{ + Addr: viper.GetString("redis.addr"), + Password: viper.GetString("redis.password"), + DB: viper.GetInt("redis.db"), + }) + if _, err := rdb.Ping(rdb.Context()).Result(); err != nil { + log.Fatalf("❌ Redis 连接失败: %v", err) + } + return rdb +} + +func main() { + initConfig() + db := initDB() + rdb := initRedis() + + ws.StartWorkerPool() + defer ws.StopWorkerPool() + service.InitChatService(db, rdb) + go turnserver.Start() + + r := gin.Default() + + // 允许跨域 (重要：为了前端本地开发) + r.Use(func(c *gin.Context) { + c.Writer.Header().Set("Access-Control-Allow-Origin", "*") + c.Writer.Header().Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE") + c.Writer.Header().Set("Access-Control-Allow-Headers", "Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization, X-User-ID") + if c.Request.Method == "OPTIONS" { + c.AbortWithStatus(204) + return + } + c.Next() + }) + + r.GET("/ws", func(c *gin.Context) { + upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }} + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + return + } + + userID := c.Query("user_id") + clientID := fmt.Sprintf("%s-%d", viper.GetString("app.node_id"), time.Now().UnixNano()) + client := &manager.Client{ID: clientID, Conn: conn, SendQueue: make(chan []byte, 256)} + + manager.Manager.Register(client) + if userID != "" { + service.ChatSvc.BindUser(client, userID) + } + client.SendQueue <- []byte(fmt.Sprintf(`{"clientId": "%s"}`, clientID)) + + for { + _, message, err := conn.ReadMessage() + if err != nil { + manager.Manager.Unregister(client) + break + } + // WebSocket 仅处理信令和心跳，不再处理 send_message (改走API) + // 但为了兼容，仍保留 PushTask + ws.PushTask(client, message) + } + }) + + apiGroup := r.Group("/api") + { + apiGroup.POST("/send", api.SendHandler) + apiGroup.POST("/send-to-user", api.SendToUserHandler) + apiGroup.POST("/bind", api.BindHandler) + apiGroup.GET("/check-user-online", api.CheckUserOnlineHandler) + apiGroup.GET("/messages", api.HistoryHandler) + apiGroup.GET("/messages/sync", api.SyncMessagesHandler) + apiGroup.GET("/health", api.HealthHandler) + apiGroup.GET("/ice-servers", api.ICEHandler) + + // 新增：联系人列表 + apiGroup.GET("/contacts", api.ContactListHandler) + } + + port := viper.GetString("app.port") + log.Printf("🚀 服务启动在端口: %s", port) + r.Run(":" + port) +} diff --git a/configs/config.yaml b/configs/config.yaml new file mode 100644 index 0000000..685cd8f --- /dev/null +++ b/configs/config.yaml @@ -0,0 +1,51 @@ +# ========================================== +# 应用基础配置 +# ========================================== +app: + name: "xk-websocket-v2" + # 服务监听端口 (HTTP & WebSocket) + port: "12080" + # 节点唯一标识符 (集群模式下必须唯一，用于路由消息) + node_id: "node-01" + # 日志级别: debug, info, warn, error + log_level: "debug" + +# ========================================== +# MySQL 数据库配置 +# ========================================== +database: + # 数据库连接字符串 (DSN) + # 格式: user:password@tcp(host:port)/dbname?charset=utf8mb4&parseTime=True&loc=Local + dsn: "root:root@tcp(127.0.0.1:3306)/nl_im?charset=utf8mb4&parseTime=True&loc=Local" + # 连接池最大空闲连接数 + max_idle_conns: 10 + # 连接池最大打开连接数 + max_open_conns: 100 + +# ========================================== +# Redis 配置 (用于缓存和集群消息广播) +# ========================================== +redis: + addr: "127.0.0.1:6379" + password: "redis_pBjaRs" + # 数据库索引 (0-15) + db: 0 + +# ========================================== +# TURN/STUN 服务器配置 (WebRTC 中继) +# ========================================== +turn: + # 是否启用内置 TURN 服务 + enabled: true + # 服务器公网 IP (重要：客户端必须能访问该IP) + public_ip: "127.0.0.1" + # TURN 服务监听端口 (UDP & TCP) + listen_port: 3478 + # 认证领域名称 + realm: "xk-chat" + # 共享密钥 (用于生成和校验临时凭证，分布式部署时所有节点必须一致) + shared_secret: "xk-super-secret-key-2025" + # UDP 端口范围下限 + min_port: 49152 + # UDP 端口范围上限 + max_port: 65535 \ No newline at end of file diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..30b6d67 --- /dev/null +++ b/go.mod @@ -0,0 +1,68 @@ +module xk-websocket-v2 + +go 1.24.1 + +require ( + github.com/gin-gonic/gin v1.11.0 + github.com/go-redis/redis/v8 v8.11.5 + github.com/gorilla/websocket v1.5.3 + github.com/panjf2000/ants/v2 v2.11.3 + github.com/pion/turn/v2 v2.1.6 + github.com/spf13/viper v1.21.0 + gorm.io/driver/mysql v1.6.0 + gorm.io/gorm v1.31.1 +) + +require ( + filippo.io/edwards25519 v1.1.0 // indirect + github.com/bytedance/gopkg v0.1.3 // indirect + github.com/bytedance/sonic v1.14.2 // indirect + github.com/bytedance/sonic/loader v0.4.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/cloudwego/base64x v0.1.6 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/fsnotify/fsnotify v1.9.0 // indirect + github.com/gabriel-vasile/mimetype v1.4.11 // indirect + github.com/gin-contrib/sse v1.1.0 // indirect + 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.28.0 // indirect + github.com/go-sql-driver/mysql v1.9.3 // indirect + github.com/go-viper/mapstructure/v2 v2.4.0 // indirect + github.com/goccy/go-json v0.10.5 // indirect + github.com/goccy/go-yaml v1.19.0 // 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.3.0 // indirect + github.com/leodido/go-urn v1.4.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/pion/dtls/v2 v2.2.12 // indirect + github.com/pion/logging v0.2.4 // indirect + github.com/pion/randutil v0.1.0 // indirect + github.com/pion/stun v0.6.1 // indirect + github.com/pion/transport/v2 v2.2.10 // indirect + github.com/pion/transport/v3 v3.1.1 // indirect + github.com/quic-go/qpack v0.6.0 // indirect + github.com/quic-go/quic-go v0.57.1 // indirect + github.com/sagikazarmark/locafero v0.12.0 // indirect + github.com/spf13/afero v1.15.0 // indirect + github.com/spf13/cast v1.10.0 // indirect + github.com/spf13/pflag v1.0.10 // indirect + github.com/subosito/gotenv v1.6.0 // indirect + github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/ugorji/go/codec v1.3.1 // indirect + github.com/wlynxg/anet v0.0.5 // indirect + go.uber.org/mock v0.6.0 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/arch v0.23.0 // indirect + golang.org/x/crypto v0.45.0 // indirect + golang.org/x/net v0.47.0 // indirect + golang.org/x/sync v0.18.0 // indirect + golang.org/x/sys v0.38.0 // indirect + golang.org/x/text v0.31.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..0d20b6c --- /dev/null +++ b/go.sum @@ -0,0 +1,222 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= +github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= +github.com/bytedance/sonic v1.14.2 h1:k1twIoe97C1DtYUo+fZQy865IuHia4PR5RPiuGPPIIE= +github.com/bytedance/sonic v1.14.2/go.mod h1:T80iDELeHiHKSc0C9tubFygiuXoGzrkjKzX2quAx980= +github.com/bytedance/sonic/loader v0.4.0 h1:olZ7lEqcxtZygCK9EKYKADnpQoYkRQxaeY2NYzevs+o= +github.com/bytedance/sonic/loader v0.4.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= +github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= +github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= +github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= +github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/gabriel-vasile/mimetype v1.4.11 h1:AQvxbp830wPhHTqc1u7nzoLT+ZFxGY7emj5DR5DYFik= +github.com/gabriel-vasile/mimetype v1.4.11/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s= +github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w= +github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM= +github.com/gin-gonic/gin v1.11.0 h1:OW/6PLjyusp2PPXtyxKHU0RbX6I/l28FTdDlae5ueWk= +github.com/gin-gonic/gin v1.11.0/go.mod h1:+iq/FyxlGzII0KHiBGjuNn4UNENUlKbGlNmc+W50Dls= +github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= +github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= +github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= +github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY= +github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY= +github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= +github.com/go-playground/validator/v10 v10.28.0 h1:Q7ibns33JjyW48gHkuFT91qX48KG0ktULL6FgHdG688= +github.com/go-playground/validator/v10 v10.28.0/go.mod h1:GoI6I1SjPBh9p7ykNE/yj3fFYbyDOpwMn5KXd+m2hUU= +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.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= +github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= +github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= +github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= +github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= +github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= +github.com/goccy/go-yaml v1.19.0 h1:EmkZ9RIsX+Uq4DYFowegAuJo8+xdX3T/2dwNPXbxEYE= +github.com/goccy/go-yaml v1.19.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +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/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= +github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= +github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE= +github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU= +github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= +github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU= +github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE= +github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs= +github.com/panjf2000/ants/v2 v2.11.3 h1:AfI0ngBoXJmYOpDh9m516vjqoUu2sLrIVgppI9TZVpg= +github.com/panjf2000/ants/v2 v2.11.3/go.mod h1:8u92CYMUc6gyvTIw8Ru7Mt7+/ESnJahz5EVtqfrilek= +github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= +github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s= +github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk= +github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE= +github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms= +github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= +github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so= +github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= +github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/stun v0.6.1 h1:8lp6YejULeHBF8NmV8e2787BogQhduZugh5PdhDyyN4= +github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/8= +github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g= +github.com/pion/transport/v2 v2.2.4/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0= +github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q= +github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E= +github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkYOM= +github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ= +github.com/pion/turn/v2 v2.1.6 h1:Xr2niVsiPTB0FPtt+yAWKFUkU1eotQbGgpTIld4x1Gc= +github.com/pion/turn/v2 v2.1.6/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.57.1 h1:25KAAR9QR8KZrCZRThWMKVAwGoiHIrNbT72ULHTuI10= +github.com/quic-go/quic-go v0.57.1/go.mod h1:ly4QBAjHA2VhdnxhojRsCUOeJwKYg+taDlos92xb1+s= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/sagikazarmark/locafero v0.12.0 h1:/NQhBAkUb4+fH1jivKHWusDYFjMOOKU88eegjfxfHb4= +github.com/sagikazarmark/locafero v0.12.0/go.mod h1:sZh36u/YSZ918v0Io+U9ogLYQJ9tLLBmM4eneO6WwsI= +github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= +github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg= +github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= +github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= +github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= +github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU= +github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= +github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= +github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= +github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= +github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY= +github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= +github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= +github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= +github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= +go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/arch v0.23.0 h1:lKF64A2jF6Zd8L0knGltUnegD62JMFBiCPBmQpToHhg= +golang.org/x/arch v0.23.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE= +golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw= +golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg= +golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= +golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns= +golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= +golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI= +golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY= +golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= +golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= +golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY= +golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= +golang.org/x/term v0.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU= +golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= +golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= +golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= +golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= +gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= +gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= +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.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg= +gorm.io/gorm v1.31.1/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs= diff --git a/internal/api/handler.go b/internal/api/handler.go new file mode 100644 index 0000000..ed8a038 --- /dev/null +++ b/internal/api/handler.go @@ -0,0 +1,209 @@ +/** + * package api + * 作用：处理 HTTP 请求接口，包括消息发送、历史记录、系统状态检查等。 + * 说明：V2 版本中消息发送主要通过 HTTP 接口进行，而非 WebSocket 直接推送。 + */ +package api + +import ( + "fmt" + "net/http" + "time" + "xk-websocket-v2/internal/manager" + "xk-websocket-v2/internal/model" + "xk-websocket-v2/internal/service" + "xk-websocket-v2/internal/turnserver" + + "github.com/gin-gonic/gin" + "github.com/spf13/viper" +) + +// ========================================== +// 消息发送相关接口 +// ========================================== + +/** + * SendHandler + * 功能：通用消息发送接口 (HTTP -> WebSocket)。 + * 路径：POST /api/send + * 逻辑：支持通过 target_client_id 或 receiver_user_id 发送。 + * 重要：必须正确处理 SenderClientID，防止多端同步时误通知发送端自己。 + */ +func SendHandler(c *gin.Context) { + var req model.SendMessageReq + // 1. 绑定并校验 JSON 参数 + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "无效的JSON参数"}) + return + } + + // 2. 获取发送者 ID (从 Header 中获取，模拟鉴权) + // 生产环境应从 JWT Token 中解析 UserID + senderID := c.GetHeader("X-User-ID") + if senderID == "" { + senderID = "system" // 默认为系统消息 + } + + // 3. [关键修复] 获取发送端的 WebSocket ClientID + // 前端在调用此接口时，必须带上自己的 socket_client_id + // 如果前端没传 (兼容旧代码)，则 fallback 到 api-gateway，这会导致发送端自己也收到"其他设备接听"通知 + clientID := req.SenderClientID + if clientID == "" { + clientID = "api-gateway" + } + + // 4. 构造临时的发送者客户端对象 + // 这个对象将传递给 Service 层，用于识别消息来源 + mockClient := &manager.Client{ + UserID: senderID, + ID: clientID, + } + + // 5. 调用核心业务逻辑处理消息 + service.ChatSvc.HandleUserMessage(mockClient, &req) + + // 6. 返回成功响应 + c.JSON(http.StatusOK, gin.H{"status": "ok"}) +} + +/** + * SendToUserHandler (V1 兼容) + * 功能：专门用于给指定用户发送消息。 + * 路径：POST /api/send-to-user + */ +func SendToUserHandler(c *gin.Context) { + // 复用 SendHandler，因为 V2 的 SendHandler 已经支持 ReceiverUserID + SendHandler(c) +} + +// ========================================== +// 用户与连接管理接口 +// ========================================== + +/** + * BindHandler + * 功能：手动绑定 ClientID 和 UserID。 + * 路径：POST /api/bind + * 场景：当 WebSocket 连接建立后，客户端通过 HTTP 接口补充用户信息。 + */ +func BindHandler(c *gin.Context) { + var req model.BindReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误"}) + return + } + // 调用服务层进行绑定 + service.ChatSvc.BindUserByClientID(req.ClientID, req.UserID) + c.JSON(http.StatusOK, gin.H{"status": "success"}) +} + +/** + * CheckUserOnlineHandler + * 功能：检查用户是否在线。 + * 路径：GET /api/check-user-online + */ +func CheckUserOnlineHandler(c *gin.Context) { + userID := c.Query("user_id") + // 调用服务层查询 Redis + isOnline := service.ChatSvc.IsUserOnline(userID) + c.JSON(http.StatusOK, gin.H{"status": "success", "result": isOnline}) +} + +// ========================================== +// 数据查询接口 +// ========================================== + +/** + * HistoryHandler + * 功能：获取历史消息记录 (分页)。 + * 路径：GET /api/messages + */ +func HistoryHandler(c *gin.Context) { + roomID := c.Query("room_id") + var msgs []model.ChatMessage + + // 简单查询最近 50 条 + // 生产环境应添加 page, page_size 参数 + result := service.ChatSvc.DB.Where("room_id = ?", roomID). + Order("created_at desc").Limit(50).Find(&msgs) + + if result.Error != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "db error"}) + return + } + c.JSON(http.StatusOK, gin.H{"data": msgs}) +} + +/** + * SyncMessagesHandler + * 功能：全量同步消息 (V1 兼容)。 + * 路径：GET /api/messages/sync + */ +func SyncMessagesHandler(c *gin.Context) { + HistoryHandler(c) +} + +/** + * ContactListHandler + * 功能：获取联系人列表 (模拟数据)。 + * 路径：GET /api/contacts + * 说明：用于前端展示登录后的好友列表。 + */ +func ContactListHandler(c *gin.Context) { + // 模拟 10 个用户数据 + users := []model.UserContact{ + {ID: "1001", Name: "张三 (我)", Avatar: "张", Desc: "Golang 专家"}, + {ID: "1002", Name: "李琦 (妻)", Avatar: "李", Desc: "在线"}, + {ID: "1003", Name: "王医生", Avatar: "医", Desc: "主任医师"}, + {ID: "1004", Name: "客服小蜜", Avatar: "客", Desc: "全天在线"}, + {ID: "1005", Name: "技术支持", Avatar: "技", Desc: "请重启试试"}, + {ID: "1006", Name: "财务小赵", Avatar: "财", Desc: "报销单请提交"}, + {ID: "1007", Name: "运维阿强", Avatar: "运", Desc: "服务器维护中"}, + {ID: "1008", Name: "测试小丽", Avatar: "测", Desc: "Bug 太多了"}, + {ID: "1009", Name: "HR", Avatar: "人", Desc: "本月考勤异常"}, + {ID: "1010", Name: "老板", Avatar: "老", Desc: "今晚开会"}, + } + c.JSON(http.StatusOK, gin.H{"data": users}) +} + +// ========================================== +// 系统与 WebRTC 接口 +// ========================================== + +/** + * HealthHandler + * 功能：服务健康检查。 + * 路径：GET /api/health + */ +func HealthHandler(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{ + "status": "ok", + "node": viper.GetString("app.node_id"), + "time": time.Now().Format(time.RFC3339), + }) +} + +/** + * ICEHandler + * 功能：获取 TURN/STUN 服务器配置及临时凭证。 + * 路径：GET /api/ice-servers + * 用途：WebRTC 前端在建立 PeerConnection 前需调用此接口。 + */ +func ICEHandler(c *gin.Context) { + userID := c.Query("user_id") + + // 生成临时凭证 (HMAC-SHA1) + username, credential := turnserver.GenerateCredentials(userID) + + // 从配置读取公网IP和端口 + ip := viper.GetString("turn.public_ip") + port := viper.GetInt("turn.listen_port") + + // 构造配置返回给前端 + cfg := model.ICEServerConfig{ + Urls: []string{fmt.Sprintf("turn:%s:%d", ip, port)}, + Username: username, + Credential: credential, + } + c.JSON(http.StatusOK, gin.H{"data": []model.ICEServerConfig{cfg}}) +} diff --git a/internal/manager/client_manager.go b/internal/manager/client_manager.go new file mode 100644 index 0000000..6b7edfe --- /dev/null +++ b/internal/manager/client_manager.go @@ -0,0 +1,225 @@ +/** + * package manager + * 作用：管理全量的 WebSocket 连接，提供线程安全的注册、注销、绑定和消息发送功能。 + * 核心：使用 sync.Map 替代传统 map + RWMutex 以提高高并发下的读写性能。 + */ +package manager + +import ( + "log" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +/** + * Client + * 结构体：代表一个 WebSocket 客户端连接会话。 + */ +type Client struct { + ID string // 客户端唯一标识 (格式: NodeID-Timestamp) + UserID string // 绑定的用户ID (未绑定时为空) + Conn *websocket.Conn // 底层 WebSocket 连接对象 + SendQueue chan []byte // 发送缓冲队列 (防止网络阻塞导致写协程卡死) +} + +/** + * ClientManager + * 结构体：连接管理器，维护 ClientID 映射和 UserID 反向索引。 + */ +type ClientManager struct { + // 客户端列表: map[string]*Client (Key: ClientID) + clients sync.Map + // 用户反向索引: map[string][]string (Key: UserID, Value: [ClientID1, ClientID2]) + userClients sync.Map +} + +// Manager 全局单例 +var Manager = &ClientManager{} + +/** + * Register + * 功能：注册新建立的 WebSocket 连接。 + * @param c *Client 客户端对象 + */ +func (m *ClientManager) Register(c *Client) { + m.clients.Store(c.ID, c) + log.Printf("✅ [ClientManager] 客户端注册: %s", c.ID) + // 启动该客户端独享的写协程 + go c.writePump() +} + +/** + * Unregister + * 功能：注销连接，清理资源和索引。 + * @param c *Client 客户端对象 + */ +func (m *ClientManager) Unregister(c *Client) { + if _, ok := m.clients.Load(c.ID); ok { + m.clients.Delete(c.ID) + close(c.SendQueue) // 关闭通道，退出 writePump + c.Conn.Close() + + // 如果已绑定用户，清理反向索引 + if c.UserID != "" { + m.removeUserClient(c.UserID, c.ID) + } + log.Printf("🔌 [ClientManager] 客户端注销: %s", c.ID) + } +} + +/** + * BindUser + * 功能：将 ClientID 与 UserID 进行绑定，支持多端登录。 + * @param clientID 客户端ID + * @param userID 用户ID + */ +func (m *ClientManager) BindUser(clientID, userID string) { + clientInterface, ok := m.clients.Load(clientID) + if !ok { + return + } + client := clientInterface.(*Client) + client.UserID = userID + + // 更新反向索引 (UserID -> []ClientID) + // 使用 LoadOrStore 初始化切片 + actual, _ := m.userClients.LoadOrStore(userID, make([]string, 0)) + ids := actual.([]string) + + // 去重逻辑：防止重复添加同一个 ClientID + for _, id := range ids { + if id == clientID { + return + } + } + + ids = append(ids, clientID) + m.userClients.Store(userID, ids) + + log.Printf("🔗 [ClientManager] 绑定成功: %s -> %s", clientID, userID) +} + +/** + * removeUserClient + * 功能：从用户的客户端列表中移除指定的 ClientID。 + */ +func (m *ClientManager) removeUserClient(userID, clientID string) { + val, ok := m.userClients.Load(userID) + if !ok { + return + } + ids := val.([]string) + newIds := make([]string, 0) + for _, id := range ids { + if id != clientID { + newIds = append(newIds, id) + } + } + + if len(newIds) == 0 { + m.userClients.Delete(userID) // 如果没有设备了，删除 Key + } else { + m.userClients.Store(userID, newIds) // 更新列表 + } +} + +/** + * SendToClient + * 功能：向指定客户端发送消息 (非阻塞模式)。 + * @param clientID 目标客户端ID + * @param message 消息字节数据 + */ +func (m *ClientManager) SendToClient(clientID string, message []byte) { + if val, ok := m.clients.Load(clientID); ok { + client := val.(*Client) + select { + case client.SendQueue <- message: + // 成功入队 + default: + log.Printf("⚠️ [ClientManager] 发送队列已满，丢弃消息: %s", clientID) + } + } +} + +/** + * SendToUser + * 功能：向指定用户的所有在线设备广播消息。 + * @param userID 目标用户ID + * @param message 消息字节数据 + */ +func (m *ClientManager) SendToUser(userID string, message []byte) { + if val, ok := m.userClients.Load(userID); ok { + ids := val.([]string) + for _, clientID := range ids { + m.SendToClient(clientID, message) + } + } +} + +/** + * GetOtherClients + * 功能：获取该用户除 excludeClientID 以外的其他所有在线客户端ID。 + * 场景：用于多端同步（如一端接听，通知其他端关闭振铃）。 + */ +func (m *ClientManager) GetOtherClients(userID, excludeClientID string) []string { + var results []string + if val, ok := m.userClients.Load(userID); ok { + ids := val.([]string) + for _, id := range ids { + if id != excludeClientID { + results = append(results, id) + } + } + } + return results +} + +/** + * writePump + * 功能：每个客户端独享的写协程，确保 WebSocket 并发写安全，并处理心跳 Ping。 + */ +func (c *Client) writePump() { + ticker := time.NewTicker(50 * time.Second) // 心跳间隔 + defer func() { + ticker.Stop() + c.Conn.Close() + }() + + for { + select { + case message, ok := <-c.SendQueue: + c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if !ok { + // 队列关闭，发送 Close 帧 + c.Conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + w, err := c.Conn.NextWriter(websocket.TextMessage) + if err != nil { + return + } + w.Write(message) + + // 优化：如果队列里堆积了多条消息，一次性写完，减少系统调用次数 + n := len(c.SendQueue) + for i := 0; i < n; i++ { + w.Write([]byte{'\n'}) // JSON消息分隔符 + w.Write(<-c.SendQueue) + } + + if err := w.Close(); err != nil { + return + } + + case <-ticker.C: + // 发送心跳 Ping + c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} diff --git a/internal/model/types.go b/internal/model/types.go new file mode 100644 index 0000000..ac6166a --- /dev/null +++ b/internal/model/types.go @@ -0,0 +1,163 @@ +/** + * package model + * 作用：定义全系统通用的数据结构（PO 与 DTO）。 + * 规范：所有字段均采用 snake_case (下划线命名) 以保持 DB 和 JSON 的高度一致性。 + */ +package model + +import "time" + +// ========================================== +// 数据库实体 (PO - Persistent Object) +// ========================================== + +/** + * ChatMessage + * 对应数据库表：chat_messages + * 作用：持久化存储聊天记录，包括文本、多媒体和信令状态。 + */ +type ChatMessage struct { + // 消息唯一标识 ID + ID uint `gorm:"primaryKey;comment:消息唯一标识ID" json:"id"` + // 房间ID，用于标识聊天室或会话 + RoomID string `gorm:"type:varchar(100);index;comment:房间ID" json:"room_id"` + // 发送者用户ID + SenderUserID string `gorm:"type:varchar(100);index;comment:发送者用户ID" json:"sender_user_id"` + // 接收者用户ID + ReceiverUserID string `gorm:"type:varchar(100);index;comment:接收者用户ID" json:"receiver_user_id"` + // 消息类型 (0:文本, 1:图片, 2:音频, 3:视频, 4:处方, 5:病例, 6:视频通话, 7:语音通话, 8:文件) + MessageType int `gorm:"type:int;comment:消息类型" json:"message_type"` + // 消息内容或信令数据 + Content string `gorm:"type:text;comment:消息内容" json:"content"` + // 存储富文本元数据（如URL预览信息的JSON字符串: {"title":"...","image":"..."}） + Extra string `gorm:"type:text;comment:扩展字段，存储JSON格式的元数据" json:"extra"` + // 通话时长(秒)，仅在通话类型的消息中有效 + Duration int `gorm:"type:int;comment:通话时长(秒)" json:"duration"` + // 关联的通话ID (用于串联信令) + CallID string `gorm:"type:varchar(100);index;comment:关联的通话ID" json:"call_id"` + // 通话状态 (invite, accepted, ended, rejected, offer, answer, candidate, hangup, etc.) + CallStatus string `gorm:"type:varchar(50);comment:通话状态" json:"call_status"` + // 创建时间，自动生成 + CreatedAt time.Time `gorm:"autoCreateTime;comment:创建时间" json:"created_at"` +} + +// ========================================== +// 交互数据传输对象 (DTO - Data Transfer Object) +// ========================================== + +/** + * WsPayload + * 作用：WebSocket 传输的最外层通用载荷。 + * 逻辑：根据 RequestType 字段决定如何解析 Data。 + */ +type WsPayload struct { + // 请求类型: "bind", "heartbeat", "send_message", "receive_message" + RequestType string `json:"request_type"` + // 具体数据载荷，结构取决于 RequestType + Data interface{} `json:"data"` +} + +/** + * SendMessageReq + * 作用：客户端发送消息或信令的请求参数结构。 + */ +type SendMessageReq struct { + // [新增] 发送端的 WebSocket ClientID，用于在多端同步时排除自己 + SenderClientID string `json:"sender_client_id,omitempty"` + + // 指定目标客户端ID (选填，用于点对点精确控制) + TargetClientID string `json:"target_client_id,omitempty"` + // 指定目标用户ID (选填，用于发送给用户的所有设备) + ReceiverUserID string `json:"receiver_user_id,omitempty"` + // 房间ID + RoomID string `json:"room_id"` + // 消息类型 + MessageType int `json:"message_type"` + // 内容 + Content string `json:"content"` + // 时长 + Duration int `json:"duration"` + + // WebRTC 信令专用字段: 通话唯一ID + CallID string `json:"call_id,omitempty"` + // WebRTC 信令专用字段: 信令状态 (invite, offer, answer, candidate, hangup) + CallStatus string `json:"call_status,omitempty"` +} + +/** + * UrlMeta + * 作用：URL 预览元数据结构 (存储在 Extra 字段中) + */ +type UrlMeta struct { + // 标题 + Title string `json:"title"` + // 描述 + Description string `json:"description"` + // 图片链接 + Image string `json:"image"` + // 原始URL + Url string `json:"url"` +} + +/** + * BindReq + * 作用：用户绑定请求参数。 + */ +type BindReq struct { + // 用户ID + UserID string `json:"user_id"` + // WebSocket握手后获得的临时ID + ClientID string `json:"client_id"` +} + +/** + * ICEServerConfig + * 作用：返回给前端的 TURN/STUN 配置信息。 + */ +type ICEServerConfig struct { + // TURN服务器地址列表 + Urls []string `json:"urls"` + // 鉴权用户名 + Username string `json:"username,omitempty"` + // 鉴权密码 + Credential string `json:"credential,omitempty"` +} + +/** + * UserContact + * 作用：模拟联系人结构 + */ +type UserContact struct { + // 用户唯一ID + ID string `json:"id"` + // 用户名称 + Name string `json:"name"` + // 用户头像URL或字符 + Avatar string `json:"avatar"` + // 用户描述或签名 + Desc string `json:"desc"` +} + +/** + * ClusterMessage + * 作用：内部通信，Redis Pub/Sub 集群消息转发结构。 + */ +type ClusterMessage struct { + // 发出该消息的源节点ID + SenderNodeID string `json:"sender_node_id"` + // 目标用户ID + TargetUserID string `json:"target_user_id"` + // 原始消息体 (WsPayload) + Payload interface{} `json:"payload"` +} + +// ========================================== +// 常量定义 +// ========================================== + +const ( + // Redis Key 前缀: 用户所在节点映射 (Hash结构或String结构) + KeyUserNodeMap = "ws:user:node:" + // Redis Channel: 集群广播频道 + ChanClusterBroadcast = "ws:cluster:broadcast" +) diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go new file mode 100644 index 0000000..4ca9339 --- /dev/null +++ b/internal/service/chat_service.go @@ -0,0 +1,355 @@ +/** + * 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" + + "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 消息请求体 + */ +func (s *ChatService) HandleUserMessage(senderClient *manager.Client, req *model.SendMessageReq) { + // 1. [关键] 拦截通话信令：处理多端同步逻辑 + // 如果收到 "accepted" (接听) 信号，说明当前设备接听了电话 + // 系统需要立即通知该用户的 *其他设备* 停止振铃并显示“已在其他设备接听” + if req.CallStatus == "accepted" { + s.NotifyOtherDevices(senderClient.UserID, senderClient.ID, req.CallID) + } + + var extraData string = "" + + // 2. [新增] URL 识别与抓取逻辑 + // 仅针对文本消息 (MessageType=0) + 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("🌐 https://www.toolsid.com/scrapers.html 抓取成功: %s", meta.Title) + } + } + } + + // 3. 消息持久化 (MySQL) + // WebRTC 的 Candidate/Offer/Answer 信令通常不需要持久化，只有聊天记录和通话记录需要 + if req.MessageType != 6 && req.MessageType != 7 { // 假设 6,7 是纯信令流 + msg := model.ChatMessage{ + RoomID: req.RoomID, + SenderUserID: senderClient.UserID, + ReceiverUserID: req.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) + } + + // 4. 构建推送消息体 (DTO) + // 将持久化后的完整对象推给前端 + pushMsg := model.WsPayload{ + RequestType: "receive_message", + Data: msg, + } + msgBytes, _ := json.Marshal(pushMsg) + + // 分发消息 (路由到目标用户) + s.DispatchMessage(req.ReceiverUserID, msgBytes) + } else { + // 信令消息直接透传，不存库 + sigMsg := model.ChatMessage{ + SenderUserID: senderClient.UserID, + ReceiverUserID: req.ReceiverUserID, + MessageType: req.MessageType, + Content: req.Content, + CallID: req.CallID, + CallStatus: req.CallStatus, + CreatedAt: time.Now(), + } + pushMsg := model.WsPayload{ + RequestType: "receive_message", + Data: sigMsg, + } + msgBytes, _ := json.Marshal(pushMsg) + s.DispatchMessage(req.ReceiverUserID, msgBytes) + } +} + +// 辅助：正则提取第一个 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)(.*?)`) + matches := titleRe.FindStringSubmatch(html) + if len(matches) > 1 { + meta.Title = strings.TrimSpace(matches[1]) + } + + // 优化 og:image 正则 + imgRe := regexp.MustCompile(`(?i) 1 { + meta.Image = imgMatches[1] + } + + // 优化 description 正则 + descRe := regexp.MustCompile(`(?i) 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) + } +} + +/** + * 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) + } +} + +/** + * 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, + }, + } +} diff --git a/internal/turnserver/server.go b/internal/turnserver/server.go new file mode 100644 index 0000000..e49973d --- /dev/null +++ b/internal/turnserver/server.go @@ -0,0 +1,120 @@ +/** + * package turnserver + * 作用：基于 pion/turn 封装内置的 WebRTC 中继服务。 + * 特性： + * 1. 支持 UDP 和 TCP 协议。 + * 2. 手动实现了长期凭证机制 (Long-Term Credential Mechanism) 进行鉴权。 + * 3. 支持分布式部署，通过共享密钥验证 Token。 + */ +package turnserver + +import ( + "crypto/hmac" + "crypto/sha1" + "encoding/base64" + "fmt" + "log" + "net" + "strconv" + "time" + + "github.com/pion/turn/v2" + "github.com/spf13/viper" +) + +/** + * Start + * 功能：初始化并启动 TURN 服务器。 + * 注意：这是一个阻塞调用，通常建议在 goroutine 中运行。 + */ +func Start() { + if !viper.GetBool("turn.enabled") { + log.Println("ℹ️ [TURN] 服务已禁用") + return + } + + publicIP := viper.GetString("turn.public_ip") + port := viper.GetInt("turn.listen_port") + realm := viper.GetString("turn.realm") + + // 监听 UDP 端口 + udpListener, err := net.ListenPacket("udp4", "0.0.0.0:"+strconv.Itoa(port)) + if err != nil { + log.Fatalf("❌ [TURN] UDP 监听失败: %v", err) + } + + // 定义鉴权回调函数 + // 每次客户端发起 Allocate 请求时触发 + authHandler := func(username, realm string, srcAddr net.Addr) ([]byte, bool) { + // 1. (可选) 验证 username 时效性 + // username 格式: timestamp:userid + // 生产环境可以解析 timestamp 检查是否过期 + + // 2. 根据共享密钥重新计算密码 (HMAC-SHA1) + secret := viper.GetString("turn.shared_secret") + password := generateHMACPassword(username, secret) + + // 3. 生成 HA1 Key + // pion/turn 的 AuthHandler 需要返回 MD5(username:realm:password) 作为 Key + // 这是 STUN/TURN 标准的完整性校验要求 + key := turn.GenerateAuthKey(username, realm, password) + return key, true + } + + // 创建 TURN Server 实例 + s, err := turn.NewServer(turn.ServerConfig{ + Realm: realm, + AuthHandler: authHandler, + // 配置中继地址生成器 (使用公网IP，否则客户端无法连接) + PacketConnConfigs: []turn.PacketConnConfig{ + { + PacketConn: udpListener, + RelayAddressGenerator: &turn.RelayAddressGeneratorStatic{ + RelayAddress: net.ParseIP(publicIP), + Address: "0.0.0.0", + }, + }, + }, + }) + + if err != nil { + log.Fatalf("❌ [TURN] 启动失败: %v", err) + } + + log.Printf("🚀 [TURN] 服务已启动 | IP: %s | Port: %d | Realm: %s", publicIP, port, realm) + + // 阻塞保持服务运行 + // 在 main.go 中，我们依靠主线程的 select 或 Web Server 阻塞，所以这里不需要 block + // 但为了代码独立性，如果是独立进程，需要阻塞。这里因为是库函数，只需返回即可，Server 会在后台运行。 + _ = s +} + +/** + * GenerateCredentials + * 功能：为特定用户生成 TURN 服务的访问凭证。 + * 算法：WebRTC 标准的 Long-Term Credential Mechanism (HMAC-SHA1)。 + * @param userID 用户ID + * @return username, password (credential) + */ +func GenerateCredentials(userID string) (string, string) { + // Username = timestamp:userid (有效期 24 小时) + timestamp := time.Now().Add(24 * time.Hour).Unix() + username := fmt.Sprintf("%d:%s", timestamp, userID) + + secret := viper.GetString("turn.shared_secret") + + // 生成密码 + password := generateHMACPassword(username, secret) + return username, password +} + +/** + * generateHMACPassword + * 功能：辅助函数，使用 HMAC-SHA1 算法生成密码 + * 标准：Base64(HMAC-SHA1(secret, username)) + */ +func generateHMACPassword(username, secret string) string { + mac := hmac.New(sha1.New, []byte(secret)) + mac.Write([]byte(username)) + return base64.StdEncoding.EncodeToString(mac.Sum(nil)) +} diff --git a/internal/ws/worker.go b/internal/ws/worker.go new file mode 100644 index 0000000..b54194b --- /dev/null +++ b/internal/ws/worker.go @@ -0,0 +1,207 @@ +/** + * package ws + * 作用：处理 WebSocket 协议层逻辑，并管理基于 ants 的协程工作池。 + * 核心：将 I/O 读取与业务逻辑分离，防止海量连接导致 Goroutine 爆炸。 + */ +package ws + +import ( + "encoding/json" + "fmt" + "log" + "runtime/debug" + + "xk-websocket-v2/internal/manager" + "xk-websocket-v2/internal/model" + "xk-websocket-v2/internal/service" + + "github.com/panjf2000/ants/v2" +) + +// ========================================== +// 1. 配置常量与全局变量 +// ========================================== + +const ( + // MaxWorkerPoolSize 定义最大的业务处理协程数 + // 即使有 100万 个连接，同时也只有 5000 个协程在处理具体的业务逻辑（发消息、存库等） + // 其余连接处于空闲或等待 I/O 状态，极大节省内存。 + MaxWorkerPoolSize = 5000 +) + +// GlobalWorkerPool 全局协程池实例 +var GlobalWorkerPool *ants.PoolWithFunc + +// ========================================== +// 2. 任务结构定义 +// ========================================== + +/** + * Job + * 结构体：代表一个具体的待处理任务 + * 从 WebSocket 读到的原始 bytes 和对应的 Client 对象打包在一起 + */ +type Job struct { + Client *manager.Client // 消息来源客户端 + Payload []byte // 原始消息数据 +} + +// ========================================== +// 3. 初始化与启动 +// ========================================== + +/** + * StartWorkerPool + * 功能：初始化 ants 协程池 + * 注意：必须在 main.go 中尽早调用 + */ +func StartWorkerPool() { + var err error + + // NewPoolWithFunc 创建一个带任务处理函数的池子 + // 参数1: 池容量 + // 参数2: 具体的任务处理函数 (消费者逻辑) + GlobalWorkerPool, err = ants.NewPoolWithFunc(MaxWorkerPoolSize, func(i interface{}) { + // 类型断言：将接口转换为具体的 Job 结构体 + job, ok := i.(Job) + if !ok { + log.Printf("❌ [Worker] 类型断言失败") + return + } + + // 执行业务逻辑 + handleMessage(job.Client, job.Payload) + }) + + if err != nil { + log.Fatalf("❌ [Worker] 启动协程池失败: %v", err) + } + + log.Printf("⚙️ [WorkerPool] 启动成功 | 容量: %d", MaxWorkerPoolSize) +} + +/** + * StopWorkerPool + * 功能：优雅关闭协程池 + */ +func StopWorkerPool() { + if GlobalWorkerPool != nil { + GlobalWorkerPool.Release() + log.Println("🛑 [WorkerPool] 已释放") + } +} + +// ========================================== +// 4. 业务处理逻辑 (消费者) +// ========================================== + +/** + * handleMessage + * 功能：核心业务路由，解析 JSON 并调用 Service 层 + * 特性：包含 Panic Recover，防止单个请求崩溃导致整个进程退出 + */ +func handleMessage(client *manager.Client, message []byte) { + // 🛡️ Panic 恢复机制：这是生产环境必须的！ + defer func() { + if r := recover(); r != nil { + log.Printf("❌ [Worker] Panic 捕获: %v\n堆栈信息: %s", r, debug.Stack()) + // 可选：发送错误日志到监控系统 (Sentry/Prometheus) + } + }() + + // 1. 基础格式校验 + if len(message) == 0 { + return + } + + // 2. 解析通用载荷 (RequestType) + var req model.WsPayload + if err := json.Unmarshal(message, &req); err != nil { + log.Printf("⚠️ [Worker] JSON解析失败: %v | 来源: %s", err, client.ID) + return + } + + // 3. 业务路由分发 + switch req.RequestType { + + case "bind": + // 处理用户绑定 (bind) + // 格式: {"request_type": "bind", "data": {"user_id": "1001"}} + var bindData model.BindReq + if err := json.Unmarshal(jsonBytes(req.Data), &bindData); err == nil { + // 调用 Service 层逻辑 + service.ChatSvc.BindUser(client, bindData.UserID) + } else { + log.Printf("⚠️ [Worker] 绑定参数错误: %v", err) + } + + case "send_message": + // 处理消息发送 (send_message) + // 包含：文本聊天、图片、视频通话信令等 + var msgData model.SendMessageReq + if err := json.Unmarshal(jsonBytes(req.Data), &msgData); err == nil { + // 核心：调用 ChatService 处理消息（持久化、转发、多端同步） + // 注意：这里传入的是 client 指针，以便 Service 层获取发送者的 UserID 和 ClientID + service.ChatSvc.HandleUserMessage(client, &msgData) + } else { + log.Printf("⚠️ [Worker] 消息参数错误: %v", err) + } + + case "heartbeat": + // 处理心跳 (heartbeat) + // 快速响应，不需要走复杂业务逻辑 + // 直接往客户端写回 Pong 消息 + pong := []byte(`{"request_type":"heartbeat_ack"}`) + manager.Manager.SendToClient(client.ID, pong) + + default: + log.Printf("❓ [Worker] 未知请求类型: %s", req.RequestType) + } +} + +// ========================================== +// 5. 辅助工具 +// ========================================== + +/** + * jsonBytes + * 功能：辅助函数，将 map[string]interface{} 或其他类型转回 []byte + * 场景：因为 WsPayload.Data 是 interface{}，在 Unmarshal 时可能被解析为 map，需要二次转换 + */ +func jsonBytes(v interface{}) []byte { + if v == nil { + return []byte("{}") + } + + // 如果已经是 string，直接转换 + if str, ok := v.(string); ok { + return []byte(str) + } + + // 否则重新序列化 + b, _ := json.Marshal(v) + return b +} + +/** + * PushTask + * 功能：对外暴露的投递函数，供 client_manager 或 main.go 调用 + * 作用：非阻塞地将任务扔进池子 + * @param client 消息来源客户端 + * @param payload 原始消息 + */ +func PushTask(client *manager.Client, payload []byte) error { + if GlobalWorkerPool == nil { + return fmt.Errorf("worker pool not initialized") + } + + // 包装任务 + job := Job{ + Client: client, + Payload: payload, + } + + // Invoke 会从池里取一个 worker 来执行 NewPoolWithFunc 注册的函数 + // 如果池满了，Invoke 会阻塞，直到有 worker 空闲 (或者返回错误，取决于配置) + return GlobalWorkerPool.Invoke(job) +}