759 lines
32 KiB
Go
759 lines
32 KiB
Go
/**
|
||
* package main
|
||
*
|
||
* IM系统后端服务主程序
|
||
*
|
||
* 功能概述：
|
||
* 1. 初始化配置、数据库、Redis等基础服务
|
||
* 2. 初始化业务服务（认证、用户、联系人、房间、聊天、附件等）
|
||
* 3. 启动WebSocket服务器和TURN服务器
|
||
* 4. 注册HTTP API路由和中间件
|
||
* 5. 启动HTTP服务器
|
||
*
|
||
* 技术栈：
|
||
* - Gin: HTTP Web框架
|
||
* - GORM: ORM数据库操作
|
||
* - Redis: 缓存和消息队列
|
||
* - WebSocket: 实时通信
|
||
* - JWT: 身份认证
|
||
*/
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"net/http"
|
||
"os"
|
||
"os/signal"
|
||
"strings"
|
||
"syscall"
|
||
"time"
|
||
|
||
"xk-websocket-v2/internal/api"
|
||
"xk-websocket-v2/internal/manager"
|
||
"xk-websocket-v2/internal/mediaserver"
|
||
"xk-websocket-v2/internal/middleware"
|
||
"xk-websocket-v2/internal/model"
|
||
"xk-websocket-v2/internal/service"
|
||
"xk-websocket-v2/internal/turnserver"
|
||
"xk-websocket-v2/internal/utils"
|
||
"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"
|
||
)
|
||
|
||
// WebSocket 读端防护参数
|
||
// wsMaxMessageBytes: 单帧最大字节数，超过直接断开，防止超大帧耗尽内存
|
||
// wsReadWait: 读超时时间，须大于 writePump 的 Ping 周期（50s），
|
||
// 收到 Pong 或业务消息时刷新；超时未收到任何数据则判定连接假死并回收
|
||
const (
|
||
wsMaxMessageBytes = 512 * 1024
|
||
wsReadWait = 90 * time.Second
|
||
)
|
||
|
||
/**
|
||
* initConfig
|
||
*
|
||
* 功能：初始化配置文件读取器
|
||
*
|
||
* 步骤：
|
||
* 1. 设置配置文件名为 "config"
|
||
* 2. 设置配置文件类型为 YAML
|
||
* 3. 添加配置文件搜索路径（configs目录和当前目录）
|
||
* 4. 读取配置文件，如果失败则终止程序
|
||
*/
|
||
// parseCSV 解析逗号分隔的字符串为去空白的切片（用于读取来源白名单等配置）
|
||
func parseCSV(s string) []string {
|
||
if strings.TrimSpace(s) == "" {
|
||
return nil
|
||
}
|
||
parts := strings.Split(s, ",")
|
||
out := make([]string, 0, len(parts))
|
||
for _, p := range parts {
|
||
if v := strings.TrimSpace(p); v != "" {
|
||
out = append(out, v)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// isOriginAllowed 校验 WebSocket 来源(Origin)是否在白名单内；白名单为空时放行（仅建议本地开发）
|
||
func isOriginAllowed(origin string, allowed []string) bool {
|
||
if len(allowed) == 0 {
|
||
return true
|
||
}
|
||
for _, a := range allowed {
|
||
if a == origin {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func initConfig() {
|
||
viper.SetConfigName("config")
|
||
viper.SetConfigType("yaml")
|
||
viper.AddConfigPath("configs")
|
||
viper.AddConfigPath(".")
|
||
// 允许用环境变量覆盖配置中的敏感项（如 JWT_SECRET、DATABASE_DSN、REDIS_PASSWORD、TURN_SHARED_SECRET）。
|
||
// 生产环境应通过环境变量注入密钥/密码，config.yaml 仅保留本地开发默认值。
|
||
viper.AutomaticEnv()
|
||
viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
|
||
if err := viper.ReadInConfig(); err != nil {
|
||
log.Fatalf("❌ 无法读取配置文件: %v", err)
|
||
}
|
||
}
|
||
|
||
// isProdEnv 判断是否为生产环境（app.env 为 production/prod），
|
||
// 口径与 internal/api/auth_handler.go 的同名函数一致（包不同无法复用，各自维护4行）。
|
||
// 用途：数据库自动迁移只允许在开发环境执行，生产环境启动必须零 DDL。
|
||
func isProdEnv() bool {
|
||
env := viper.GetString("app.env")
|
||
return env == "production" || env == "prod"
|
||
}
|
||
|
||
/**
|
||
* initDB
|
||
*
|
||
* 功能：初始化MySQL数据库连接并执行数据库迁移
|
||
*
|
||
* 步骤：
|
||
* 1. 从配置文件中读取数据库连接字符串（DSN）
|
||
* 2. 使用GORM连接MySQL数据库
|
||
* 3. 如果连接失败，终止程序
|
||
* 4. 仅开发环境：执行 schema 自愈 + 自动迁移 + 表注释（生产环境启动零 DDL）
|
||
* 5. 返回数据库连接实例
|
||
*
|
||
* @returns *gorm.DB 数据库连接实例
|
||
*/
|
||
func initDB() *gorm.DB {
|
||
// 步骤1: 从配置文件读取数据库连接字符串
|
||
dsn := viper.GetString("database.dsn")
|
||
|
||
// 步骤2: 使用GORM连接MySQL数据库
|
||
// DisableForeignKeyConstraintWhenMigrating: 禁止 AutoMigrate 自动创建外键约束。
|
||
// 原因：chat_conversations.target_id 既可能是用户ID也可能是群ID，历史上 GORM
|
||
// 自动建的外键导致群会话插入失败，代码只能靠 SET FOREIGN_KEY_CHECKS=0 绕过（已移除）。
|
||
// 表结构与约束统一以 nl_im_plus.sql / migrations 下的 SQL 脚本为准。
|
||
db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
|
||
DisableForeignKeyConstraintWhenMigrating: true,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("❌ 数据库连接失败: %v", err)
|
||
}
|
||
|
||
// 步骤2.5: 应用连接池配置。
|
||
// config.yaml 中声明了 max_idle_conns/max_open_conns，但此前从未真正设置到
|
||
// 底层 sql.DB 上（配置形同虚设，走的是驱动默认值），高并发下连接数不可控
|
||
sqlDB, err := db.DB()
|
||
if err != nil {
|
||
log.Fatalf("❌ 获取底层数据库连接失败: %v", err)
|
||
}
|
||
maxIdleConns := viper.GetInt("database.max_idle_conns")
|
||
if maxIdleConns <= 0 {
|
||
maxIdleConns = 10
|
||
}
|
||
maxOpenConns := viper.GetInt("database.max_open_conns")
|
||
if maxOpenConns <= 0 {
|
||
maxOpenConns = 100
|
||
}
|
||
sqlDB.SetMaxIdleConns(maxIdleConns)
|
||
sqlDB.SetMaxOpenConns(maxOpenConns)
|
||
// 连接最长存活1小时：避免被 MySQL wait_timeout 掐断后拿到失效连接
|
||
sqlDB.SetConnMaxLifetime(time.Hour)
|
||
|
||
// 步骤2.6: 环境闸门——数据库迁移（ensureSchema/AutoMigrate/表注释，均含 DDL）只在开发环境执行。
|
||
// 为什么：打包部署的生产环境启动绝不允许改表结构——多节点同时启动会并发迁移互相冲突、
|
||
// 大表 DDL 锁表可能拖垮线上服务、模型与线上库的历史差异还可能触发误改。
|
||
// 生产库结构变更统一由 DBA 手动执行 migrations/ 下的 SQL 脚本（01~07 均幂等可重复执行）。
|
||
if isProdEnv() {
|
||
log.Println("🔒 生产环境跳过数据库自动迁移，表结构以 migrations/ SQL 脚本为准")
|
||
return db
|
||
}
|
||
|
||
// 步骤2.7: schema 自愈（仅开发环境）——把 migrations/ 下需人工执行的结构变更在启动时幂等补齐。
|
||
// 为什么必须在 AutoMigrate 之前：moment_likes 的历史重复数据不先清理，
|
||
// AutoMigrate 建 uk_moment_user 唯一索引会直接失败导致服务启不来
|
||
if err := ensureSchema(db); err != nil {
|
||
log.Fatalf("❌ 数据库结构自愈失败: %v", err)
|
||
}
|
||
|
||
// 步骤3: 执行自动数据库迁移，创建所有表结构
|
||
err = db.AutoMigrate(
|
||
&model.ChatMessage{},
|
||
&model.User{},
|
||
&model.UserContact{},
|
||
&model.ContactGroup{},
|
||
&model.ChatRoom{},
|
||
&model.ChatConversation{},
|
||
&model.FriendRequest{},
|
||
&model.VerificationCode{},
|
||
&model.Attachment{},
|
||
&model.ApiRequestLog{},
|
||
&model.LoginLog{},
|
||
&model.Moment{},
|
||
&model.MomentLike{},
|
||
&model.MomentComment{},
|
||
&model.MomentNotification{},
|
||
&model.MessageReadReceipt{},
|
||
&model.UserSetting{},
|
||
// AI 机器人两张表（等价 migrations/07）：此前遗漏，漏执行脚本的开发库 /api/ai/* 会运行时报错;
|
||
// users.is_bot 列已随 User 模型由上方 AutoMigrate 自动补齐
|
||
&model.AIConfig{},
|
||
&model.AIBot{},
|
||
)
|
||
if err != nil {
|
||
log.Fatalf("❌ 数据库迁移失败: %v", err)
|
||
}
|
||
|
||
// 步骤4: 为所有表添加中文注释，提高数据库可读性
|
||
tableComments := []struct {
|
||
table string
|
||
comment string
|
||
}{
|
||
{"chat_messages", "聊天消息表，持久化存储聊天记录，包括文本、多媒体和信令状态"},
|
||
{"users", "用户基本信息表"},
|
||
{"user_contacts", "用户联系人表，存储好友关系、分组、备注等信息"},
|
||
{"chat_conversations", "用户会话表，记录每个用户的最近会话列表和未读信息"},
|
||
{"contact_groups", "联系人分组表"},
|
||
{"chat_rooms", "聊天房间表，支持点对点和群聊"},
|
||
{"friend_requests", "好友申请表"},
|
||
{"verification_codes", "验证码表"},
|
||
{"attachments", "附件表，记录上传的文件信息"},
|
||
{"api_request_logs", "接口请求日志表"},
|
||
{"login_logs", "登录日志表"},
|
||
{"moments", "朋友圈动态表"},
|
||
{"moment_likes", "动态点赞表"},
|
||
{"moment_comments", "动态评论表"},
|
||
{"moment_notifications", "朋友圈通知表"},
|
||
{"message_read_receipts", "消息已读回执表"},
|
||
{"user_settings", "用户个性化设置表"},
|
||
{"room_members", "房间成员表，存储房间与用户的关联关系"},
|
||
{"message_deletions", "消息删除记录表，记录哪个用户删除了哪条消息(仅影响本人可见性)"},
|
||
{"ai_configs", "AI提供商配置表(全局一条,仅管理员维护)"},
|
||
{"ai_bots", "AI机器人定义表"},
|
||
}
|
||
|
||
for _, tc := range tableComments {
|
||
sql := fmt.Sprintf("ALTER TABLE `%s` COMMENT = '%s'", tc.table, tc.comment)
|
||
if err := db.Exec(sql).Error; err != nil {
|
||
log.Printf("⚠️ 添加表注释失败 %s: %v", tc.table, err)
|
||
}
|
||
}
|
||
|
||
log.Println("✅ 数据库迁移完成")
|
||
return db
|
||
}
|
||
|
||
/**
|
||
* ensureSchema
|
||
*
|
||
* 功能：启动时的数据库结构自愈——把 migrations/ 下需人工执行的变更以幂等 SQL 补齐。
|
||
* 注意：仅开发环境执行（initDB 里被 isProdEnv 闸门拦截），生产环境启动零 DDL，
|
||
* 线上结构变更必须由 DBA 手动执行 migrations/ 脚本。
|
||
*
|
||
* 为什么需要：项目约定表结构以 SQL 脚本为准（新表/新列不走 GORM 模型迁移），
|
||
* 但脚本依赖人工执行，漏执行的开发库会出现"代码引用了不存在的表/列"的运行时错误，
|
||
* 甚至 AutoMigrate 因存量脏数据建唯一索引失败而直接启动崩溃。
|
||
* 这里在启动路径上自动补齐，等价脚本（03/04/05）保留供 DBA 手动可控升级，
|
||
* 已手动执行过的环境下各步骤经 information_schema / IF NOT EXISTS 判断后跳过。
|
||
*
|
||
* 自愈内容：
|
||
* 1. room_members 表（等价 migrations/02+03）：建表含 nickname 列；老表缺列则补列
|
||
* 2. message_deletions 表（等价 migrations/04）：删除仅对我生效的记录表
|
||
* 3. moment_likes 唯一索引（等价 migrations/05）：先清理重复点赞、修正 like_count，
|
||
* 再建 uk_moment_user——否则 AutoMigrate 在有重复数据的库上建索引必失败
|
||
*/
|
||
func ensureSchema(db *gorm.DB) error {
|
||
// ---- 1. room_members：建表（含 nickname）或给老表补 nickname 列 ----
|
||
// 不显式指定 COLLATE：沿用库默认排序规则，与 AutoMigrate 建的其他表保持一致，
|
||
// 避免 JOIN users 等表时出现 "Illegal mix of collations"
|
||
if err := db.Exec("CREATE TABLE IF NOT EXISTS `room_members` (" +
|
||
"`room_id` varchar(100) NOT NULL COMMENT '房间ID'," +
|
||
"`user_id` varchar(100) NOT NULL COMMENT '用户ID'," +
|
||
"`role` tinyint(1) NOT NULL DEFAULT 0 COMMENT '成员角色(0=成员,1=管理员,2=群主)'," +
|
||
"`joined_at` datetime(3) NULL DEFAULT NULL COMMENT '加入时间'," +
|
||
"`muted_until` datetime NULL DEFAULT NULL COMMENT '禁言到期时间'," +
|
||
"`nickname` varchar(100) NOT NULL DEFAULT '' COMMENT '群名片(群内显示昵称)'," +
|
||
"PRIMARY KEY (`room_id`, `user_id`)," +
|
||
"INDEX `idx_room_members_user_id`(`user_id`)" +
|
||
") ENGINE = InnoDB CHARACTER SET = utf8mb4 COMMENT = '房间成员表，存储房间与用户的关联关系'").Error; err != nil {
|
||
return fmt.Errorf("创建 room_members 表失败: %w", err)
|
||
}
|
||
// 表已存在的老库：检查 nickname 列，缺则补（MySQL 无 ADD COLUMN IF NOT EXISTS）
|
||
var nickCount int64
|
||
if err := db.Raw("SELECT COUNT(*) FROM information_schema.COLUMNS " +
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'room_members' AND COLUMN_NAME = 'nickname'").
|
||
Scan(&nickCount).Error; err != nil {
|
||
return fmt.Errorf("检查 room_members.nickname 列失败: %w", err)
|
||
}
|
||
if nickCount == 0 {
|
||
if err := db.Exec("ALTER TABLE `room_members` ADD COLUMN `nickname` varchar(100) NOT NULL DEFAULT '' " +
|
||
"COMMENT '群名片(群内显示昵称)' AFTER `muted_until`").Error; err != nil {
|
||
return fmt.Errorf("补充 room_members.nickname 列失败: %w", err)
|
||
}
|
||
log.Println("🔧 [Schema] room_members 已补充 nickname 列")
|
||
}
|
||
|
||
// ---- 2. message_deletions：消息删除记录表（删除仅对我生效） ----
|
||
if err := db.Exec("CREATE TABLE IF NOT EXISTS `message_deletions` (" +
|
||
"`id` bigint UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键'," +
|
||
"`message_id` bigint UNSIGNED NOT NULL COMMENT '被删除的消息ID'," +
|
||
"`user_id` varchar(100) NOT NULL COMMENT '执行删除的用户ID(删除仅对该用户生效)'," +
|
||
"`created_at` datetime(3) NULL DEFAULT NULL COMMENT '删除时间'," +
|
||
"PRIMARY KEY (`id`)," +
|
||
// 唯一索引防止同一用户对同一消息重复插入删除记录
|
||
"UNIQUE INDEX `uk_message_user`(`message_id`, `user_id`)," +
|
||
"INDEX `idx_message_deletions_user_id`(`user_id`)" +
|
||
") ENGINE = InnoDB CHARACTER SET = utf8mb4 COMMENT = '消息删除记录表，记录哪个用户删除了哪条消息(仅影响本人可见性)'").Error; err != nil {
|
||
return fmt.Errorf("创建 message_deletions 表失败: %w", err)
|
||
}
|
||
|
||
// ---- 3. moment_likes：清理重复点赞后补 uk_moment_user 唯一索引 ----
|
||
// 仅处理"表已存在且索引缺失"的存量库；全新环境表还不存在，
|
||
// 跳过后由 AutoMigrate 直接建表+索引（空表无重复数据，安全）
|
||
var likeTableCount int64
|
||
if err := db.Raw("SELECT COUNT(*) FROM information_schema.TABLES " +
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'moment_likes'").
|
||
Scan(&likeTableCount).Error; err != nil {
|
||
return fmt.Errorf("检查 moment_likes 表失败: %w", err)
|
||
}
|
||
if likeTableCount > 0 {
|
||
var idxCount int64
|
||
if err := db.Raw("SELECT COUNT(*) FROM information_schema.STATISTICS " +
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'moment_likes' AND INDEX_NAME = 'uk_moment_user'").
|
||
Scan(&idxCount).Error; err != nil {
|
||
return fmt.Errorf("检查 moment_likes 唯一索引失败: %w", err)
|
||
}
|
||
if idxCount == 0 {
|
||
// 3.1 清理历史重复数据（保留每组最早一条），否则建唯一索引会失败
|
||
if err := db.Exec("DELETE ml FROM `moment_likes` ml " +
|
||
"INNER JOIN `moment_likes` ml2 ON ml.moment_id = ml2.moment_id " +
|
||
"AND ml.user_id = ml2.user_id AND ml.id > ml2.id").Error; err != nil {
|
||
return fmt.Errorf("清理 moment_likes 重复数据失败: %w", err)
|
||
}
|
||
// 3.2 修正因重复点赞虚高的 like_count（以实际点赞记录数为准）
|
||
if err := db.Exec("UPDATE `moments` m SET m.like_count = " +
|
||
"(SELECT COUNT(*) FROM `moment_likes` ml WHERE ml.moment_id = m.id)").Error; err != nil {
|
||
return fmt.Errorf("修正 moments.like_count 失败: %w", err)
|
||
}
|
||
// 3.3 建唯一索引：并发点赞由数据库兜底，代码层把唯一冲突当"已点赞"幂等处理
|
||
if err := db.Exec("ALTER TABLE `moment_likes` ADD UNIQUE INDEX `uk_moment_user`(`moment_id`, `user_id`)").Error; err != nil {
|
||
return fmt.Errorf("创建 moment_likes 唯一索引失败: %w", err)
|
||
}
|
||
log.Println("🔧 [Schema] moment_likes 已去重并补充 uk_moment_user 唯一索引")
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
/**
|
||
* initRedis
|
||
*
|
||
* 功能：初始化Redis连接
|
||
*
|
||
* 步骤：
|
||
* 1. 从配置文件读取Redis连接信息（地址、密码、数据库编号）
|
||
* 2. 创建Redis客户端实例
|
||
* 3. 执行Ping操作测试连接
|
||
* 4. 如果连接失败，终止程序
|
||
* 5. 返回Redis客户端实例
|
||
*
|
||
* @returns *redis.Client Redis客户端实例
|
||
*/
|
||
func initRedis() *redis.Client {
|
||
// 步骤1: 从配置文件读取Redis连接信息
|
||
rdb := redis.NewClient(&redis.Options{
|
||
Addr: viper.GetString("redis.addr"), // Redis服务器地址
|
||
Password: viper.GetString("redis.password"), // Redis密码
|
||
DB: viper.GetInt("redis.db"), // Redis数据库编号
|
||
})
|
||
|
||
// 步骤2: 执行Ping操作测试连接是否正常
|
||
if _, err := rdb.Ping(rdb.Context()).Result(); err != nil {
|
||
log.Fatalf("❌ Redis 连接失败: %v", err)
|
||
}
|
||
|
||
return rdb
|
||
}
|
||
|
||
/**
|
||
* main
|
||
*
|
||
* 功能：程序主入口，初始化所有服务并启动HTTP服务器
|
||
*
|
||
* 执行流程：
|
||
* 1. 初始化配置、数据库、Redis
|
||
* 2. 启动WebSocket工作池
|
||
* 3. 初始化雪花ID生成器
|
||
* 4. 初始化所有业务服务
|
||
* 5. 启动TURN服务器（用于WebRTC）
|
||
* 6. 创建Gin路由引擎
|
||
* 7. 注册中间件（响应时间、请求日志、CORS）
|
||
* 8. 注册WebSocket路由
|
||
* 9. 注册HTTP API路由
|
||
* 10. 启动HTTP服务器
|
||
*/
|
||
func main() {
|
||
// 步骤1: 初始化基础服务
|
||
initConfig() // 读取配置文件
|
||
// 配置加载完成后再初始化 JWT 密钥，确保 config/env 中的 jwt.secret 真正生效
|
||
// （历史问题：utils 包级 init 早于配置读取，导致永远使用默认密钥）
|
||
utils.InitJWT(viper.GetString("jwt.secret"))
|
||
db := initDB() // 连接MySQL数据库
|
||
rdb := initRedis() // 连接Redis
|
||
|
||
// 初始化 ip2region 离线 IP 库
|
||
ip2regionPath := viper.GetString("ip2region.db_path")
|
||
if ip2regionPath == "" {
|
||
ip2regionPath = "./ip2region.xdb"
|
||
}
|
||
if err := utils.InitIP2Region(ip2regionPath); err != nil {
|
||
log.Printf("⚠️ ip2region 初始化失败: %v", err)
|
||
}
|
||
|
||
// 步骤2: 启动WebSocket工作池，用于处理WebSocket消息
|
||
ws.StartWorkerPool()
|
||
defer ws.StopWorkerPool() // 程序退出时关闭工作池
|
||
|
||
// 步骤3: 初始化雪花ID生成器（用于生成全局唯一ID）
|
||
// 从配置文件读取数据中心ID和机器ID，如果未配置则使用默认值
|
||
datacenterID := viper.GetInt64("snowflake.datacenter_id")
|
||
if datacenterID == 0 {
|
||
datacenterID = 1 // 默认数据中心ID为1
|
||
}
|
||
machineID := viper.GetInt64("snowflake.machine_id")
|
||
if machineID == 0 {
|
||
machineID = 1 // 默认机器ID为1
|
||
}
|
||
if err := utils.InitSnowflake(datacenterID, machineID); err != nil {
|
||
log.Fatalf("❌ 初始化雪花ID生成器失败: %v", err)
|
||
}
|
||
|
||
// 步骤4: 初始化所有业务服务
|
||
service.InitChatService(db, rdb) // 聊天服务（消息处理、WebSocket分发）
|
||
service.InitAuthService(db, rdb) // 认证服务（登录、注册、验证码）
|
||
service.InitUserService(db) // 用户服务（用户信息管理）
|
||
service.InitContactService(db) // 联系人服务（好友管理、分组管理）
|
||
service.InitConversationService(db) // 会话服务（最近聊天列表）
|
||
service.InitRoomService(db) // 房间服务（聊天房间管理）
|
||
service.InitAttachmentService(db) // 附件服务（文件上传、管理）
|
||
service.InitLoginLogService(db) // 登录日志服务（记录登录历史）
|
||
service.InitMomentService(db) // 朋友圈服务（动态、点赞、评论）
|
||
service.InitSearchService(db) // 搜索服务（聚合搜索）
|
||
service.InitQRCodeLoginService(rdb) // 扫码登录服务（App 扫码登录 PC，Redis 状态机）
|
||
service.InitAIBotService(db) // AI机器人服务（工厂模式接入大模型，机器人聊天应答）
|
||
|
||
// 步骤5: 启动TURN服务器（用于WebRTC音视频通话）
|
||
go turnserver.Start()
|
||
|
||
// 步骤5.5: 启动媒体服务器（WebRTC SFU + RTMP）
|
||
go mediaserver.Start()
|
||
|
||
// 步骤6: 创建Gin路由引擎
|
||
r := gin.Default()
|
||
|
||
// 步骤7: 注册中间件（按顺序执行）
|
||
// 响应时间统计中间件（必须在最前面，用于记录请求开始时间）
|
||
r.Use(middleware.ResponseTimeMiddleware())
|
||
|
||
// 接口请求日志中间件（记录所有API请求信息）
|
||
requestLogMiddleware := middleware.NewRequestLogMiddleware(db)
|
||
r.Use(requestLogMiddleware.Handler())
|
||
|
||
// CORS跨域中间件（允许前端跨域访问，重要：为了前端本地开发）
|
||
r.Use(func(c *gin.Context) {
|
||
// 设置允许的源（*表示允许所有源）
|
||
c.Writer.Header().Set("Access-Control-Allow-Origin", "*")
|
||
// 设置允许的HTTP方法（路由中大量使用 PUT/DELETE/PATCH，如删好友、改群设置，
|
||
// 缺失会导致浏览器预检失败、相关功能在跨域场景下全部不可用）
|
||
c.Writer.Header().Set("Access-Control-Allow-Methods", "POST, GET, PUT, DELETE, PATCH, OPTIONS")
|
||
// 设置允许的请求头（X-User-ID 已废弃：身份一律取自 JWT，不再信任该头）
|
||
c.Writer.Header().Set("Access-Control-Allow-Headers", "Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization")
|
||
// 处理OPTIONS预检请求
|
||
if c.Request.Method == "OPTIONS" {
|
||
c.AbortWithStatus(204) // 返回204 No Content
|
||
return
|
||
}
|
||
c.Next()
|
||
})
|
||
|
||
// 静态文件服务（提供附件访问功能）
|
||
// 访问路径：/uploads/xxx -> ./uploads/xxx
|
||
r.Static("/uploads", "./uploads")
|
||
|
||
// PC 桌面端（Electron）更新包托管目录
|
||
// electron-updater 会到 /updates/latest.yml 检查版本，并下载同目录下的安装包与 .blockmap（增量更新）
|
||
// 发版流程：electron-builder 打包后，把 release/ 下的 latest.yml、NL-IM-Setup-x.y.z.exe、.blockmap 上传到 ./updates
|
||
r.Static("/updates", "./updates")
|
||
|
||
// 步骤8: 注册WebSocket路由
|
||
// 路径：GET /ws?token=xxx（用户身份以 JWT 为准）
|
||
// WebSocket 允许来源白名单（为空表示不限制，仅建议本地开发使用）
|
||
wsAllowedOrigins := parseCSV(viper.GetString("app.ws_allowed_origins"))
|
||
r.GET("/ws", func(c *gin.Context) {
|
||
// 步骤1: 握手鉴权——必须携带有效 JWT（?token= 或 Authorization），
|
||
// 用户身份以 Token 为准，禁止用 user_id 查询参数伪造身份（原实现无鉴权可冒充任意人）
|
||
token := c.Query("token")
|
||
if token == "" {
|
||
token = c.GetHeader("Authorization")
|
||
}
|
||
if len(token) > 7 && token[:7] == "Bearer " {
|
||
token = token[7:]
|
||
}
|
||
userID, err := utils.ValidateToken(token)
|
||
if err != nil || userID == "" {
|
||
// 统一响应结构：保持 HTTP 401（浏览器 WebSocket 握手失败只看状态码），
|
||
// 响应体复用与业务接口一致的 ApiResponse 格式，不再返回裸 gin.H
|
||
middleware.AbortUnauthorized(c, "WebSocket 认证失败")
|
||
return
|
||
}
|
||
|
||
// 步骤2: 创建WebSocket升级器，按白名单校验来源(Origin)
|
||
upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool {
|
||
return isOriginAllowed(r.Header.Get("Origin"), wsAllowedOrigins)
|
||
}}
|
||
|
||
// 步骤3: 将HTTP连接升级为WebSocket连接
|
||
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
||
if err != nil {
|
||
return // 升级失败，直接返回
|
||
}
|
||
|
||
// 步骤3.1: 读端加固
|
||
// 1) SetReadLimit 限制单帧大小，防止恶意客户端发送超大帧一次性撑爆内存；
|
||
// 2) SetReadDeadline + SetPongHandler 组成读超时机制：writePump 每 50s 发一次 Ping，
|
||
// 正常客户端自动回 Pong 会刷新读截止时间；而"半开"连接（TCP 未断、应用层无响应）
|
||
// 收不到任何数据/Pong，90s 后 ReadMessage 会超时返回，连接及其 goroutine/缓冲得以及时回收，
|
||
// 避免假死连接长期堆积。
|
||
conn.SetReadLimit(wsMaxMessageBytes)
|
||
_ = conn.SetReadDeadline(time.Now().Add(wsReadWait))
|
||
conn.SetPongHandler(func(string) error {
|
||
return conn.SetReadDeadline(time.Now().Add(wsReadWait))
|
||
})
|
||
|
||
// 步骤4: 生成唯一的客户端ID（节点ID + 时间戳）
|
||
clientID := fmt.Sprintf("%s-%d", viper.GetString("app.node_id"), time.Now().UnixNano())
|
||
|
||
// 步骤5: 创建客户端对象，包含连接和发送队列
|
||
client := &manager.Client{ID: clientID, RemoteIP: c.ClientIP(), Conn: conn, SendQueue: make(chan []byte, 256)}
|
||
|
||
// 步骤6: 注册客户端到管理器
|
||
manager.Manager.Register(client)
|
||
|
||
// 步骤7: 用 Token 中解析出的用户ID 绑定客户端（忽略查询参数中的 user_id）
|
||
service.ChatSvc.BindUser(client, userID)
|
||
|
||
// 步骤8: 发送客户端ID给前端
|
||
client.SendQueue <- []byte(fmt.Sprintf(`{"clientId": "%s"}`, clientID))
|
||
|
||
// 步骤9: 循环读取WebSocket消息
|
||
for {
|
||
_, message, err := conn.ReadMessage()
|
||
if err != nil {
|
||
// 连接断开：注销客户端并清理 Redis 路由（该用户在本节点最后一个连接下线时）
|
||
service.ChatSvc.HandleClientOffline(client)
|
||
break
|
||
}
|
||
// 收到业务消息同样视为连接存活，刷新读截止时间
|
||
_ = conn.SetReadDeadline(time.Now().Add(wsReadWait))
|
||
// WebSocket 仅处理信令和心跳，不再处理 send_message (改走API)
|
||
// 但为了兼容，仍保留 PushTask
|
||
// 协程池满时 Invoke 会返回错误，记录日志避免消息静默丢失（无观测）
|
||
if err := ws.PushTask(client, message); err != nil {
|
||
log.Printf("⚠️ [WS] 任务入池失败（协程池可能已满）: client=%s err=%v", client.ID, err)
|
||
}
|
||
}
|
||
})
|
||
|
||
// WebSocket 推流路由（不经过 authGroup，避免中间件干扰 WebSocket 升级）
|
||
// 路径：GET /api/call/ws-push?stream_id=xxx&user_id=xxx&room_id=xxx&token=xxx
|
||
r.GET("/api/call/ws-push", api.WSPushHandler)
|
||
|
||
// 步骤9: 注册HTTP API路由
|
||
apiGroup := r.Group("/api")
|
||
{
|
||
// 公开接口（不需要认证，任何人都可以访问）
|
||
apiGroup.POST("/login", api.LoginHandler)
|
||
apiGroup.POST("/register", api.RegisterHandler)
|
||
apiGroup.POST("/send-email-code", api.SendEmailCodeHandler)
|
||
apiGroup.POST("/send-sms-code", api.SendSmsCodeHandler)
|
||
apiGroup.GET("/check-token", api.CheckTokenHandler)
|
||
apiGroup.GET("/health", api.HealthHandler)
|
||
|
||
// 扫码登录（PC 侧，未登录即可访问：生成二维码 + 轮询状态）
|
||
apiGroup.POST("/qrcode/generate", api.GenerateQRCodeHandler)
|
||
apiGroup.GET("/qrcode/status", api.QRCodeStatusHandler)
|
||
|
||
// 需要认证的接口组（必须携带有效的JWT Token）
|
||
authGroup := apiGroup.Group("")
|
||
authGroup.Use(middleware.JWTAuthMiddleware()) // 使用JWT认证中间件
|
||
{
|
||
// 聚合搜索
|
||
authGroup.GET("/search", api.GlobalSearchHandler)
|
||
|
||
// 消息发送 / 绑定 / 历史（强制 JWT：发送者与绑定用户以 Token 为准，禁止伪造身份；
|
||
// 历史消息仅房间成员可读）。原先放在公开组且靠 X-User-ID 兜底，存在严重越权风险，已收敛到此。
|
||
authGroup.POST("/send", api.SendHandler)
|
||
authGroup.POST("/send-to-user", api.SendToUserHandler)
|
||
authGroup.POST("/bind", api.BindHandler)
|
||
authGroup.GET("/check-user-online", api.CheckUserOnlineHandler)
|
||
authGroup.GET("/messages", api.HistoryHandler)
|
||
authGroup.GET("/messages/sync", api.SyncMessagesHandler)
|
||
// TURN/STUN 临时凭证：需登录，凭证绑定当前用户
|
||
authGroup.GET("/ice-servers", api.ICEHandler)
|
||
|
||
// 用户管理
|
||
authGroup.GET("/user/my-info", api.GetMyInfoHandler)
|
||
authGroup.GET("/user/list", api.GetUserListHandler)
|
||
authGroup.GET("/user/:id", api.GetUserByIDHandler)
|
||
authGroup.POST("/user/create", api.CreateUserHandler)
|
||
authGroup.POST("/user/update", api.UpdateUserHandler)
|
||
authGroup.POST("/user/delete", api.DeleteUserHandler)
|
||
|
||
// 联系人管理
|
||
authGroup.GET("/contacts", api.ContactListHandler)
|
||
authGroup.GET("/contacts/search", api.SearchUsersHandler)
|
||
authGroup.POST("/contacts/add-friend", api.AddFriendHandler)
|
||
authGroup.GET("/contacts/friend-requests", api.GetFriendRequestsHandler)
|
||
authGroup.POST("/contacts/accept-request", api.AcceptFriendRequestHandler)
|
||
authGroup.POST("/contacts/reject-request", api.RejectFriendRequestHandler)
|
||
authGroup.GET("/contacts/groups", api.GetGroupsHandler)
|
||
authGroup.POST("/contacts/groups", api.CreateGroupHandler)
|
||
authGroup.POST("/contacts/groups/update/:id", api.UpdateGroupHandler)
|
||
authGroup.POST("/contacts/groups/delete/:id", api.DeleteGroupHandler)
|
||
authGroup.GET("/contacts/:id", api.GetContactDetailHandler)
|
||
authGroup.POST("/contacts/update/:id", api.UpdateContactHandler)
|
||
authGroup.POST("/contacts/delete/:id", api.DeleteContactHandler)
|
||
|
||
// 消息撤回、删除、登出、已读回执
|
||
authGroup.POST("/messages/recall", api.RecallMessageHandler)
|
||
authGroup.POST("/messages/delete", api.DeleteMessageHandler)
|
||
authGroup.POST("/messages/read-receipts", api.MarkMessagesReadHandler)
|
||
authGroup.POST("/logout", api.LogoutHandler)
|
||
|
||
// 用户设置
|
||
authGroup.GET("/user/settings", api.GetUserSettingsHandler)
|
||
authGroup.POST("/user/settings", api.UpdateUserSettingsHandler)
|
||
|
||
// 会话管理
|
||
authGroup.GET("/conversations/by-room/:room_id", api.GetConversationByRoomHandler)
|
||
authGroup.GET("/conversations", api.GetConversationListHandler)
|
||
authGroup.POST("/conversations/reset-unread", api.ResetConversationUnreadHandler)
|
||
authGroup.POST("/conversations/update", api.UpdateConversationHandler)
|
||
authGroup.POST("/conversations/delete", api.DeleteConversationHandler)
|
||
|
||
// 房间 / 群聊管理
|
||
authGroup.POST("/rooms", api.CreateRoomHandler)
|
||
authGroup.GET("/rooms/:id", api.GetRoomHandler)
|
||
|
||
// 群聊相关接口
|
||
authGroup.POST("/groups", api.CreateChatGroupHandler)
|
||
authGroup.GET("/groups", api.ListUserGroupsHandler) // 获取用户群聊列表（必须在 /groups/:room_id 之前）
|
||
authGroup.GET("/groups/:room_id", api.GetGroupInfoHandler)
|
||
authGroup.GET("/groups/:room_id/members", api.ListGroupMembersHandler)
|
||
authGroup.POST("/groups/:room_id/members", api.AddGroupMembersHandler)
|
||
authGroup.POST("/groups/:room_id/members/:user_id/remove", api.RemoveGroupMemberHandler)
|
||
authGroup.POST("/groups/:room_id/update", api.UpdateGroupInfoHandler)
|
||
authGroup.POST("/groups/:room_id/members/:user_id/role", api.ChangeMemberRoleHandler)
|
||
authGroup.POST("/groups/:room_id/quit", api.QuitGroupHandler)
|
||
authGroup.POST("/groups/:room_id/dissolve", api.DissolveGroupHandler)
|
||
authGroup.GET("/groups/:room_id/announcement", api.GetGroupAnnouncementHandler)
|
||
authGroup.POST("/groups/:room_id/announcement", api.UpdateGroupAnnouncementHandler)
|
||
authGroup.POST("/groups/:room_id/members/:user_id/mute", api.MuteGroupMemberHandler)
|
||
authGroup.GET("/groups/:room_id/members/:user_id/mute", api.GetMemberMuteStatusHandler)
|
||
authGroup.GET("/groups/:room_id/settings", api.GetGroupSettingsHandler)
|
||
authGroup.POST("/groups/:room_id/settings", api.UpdateGroupSettingsHandler)
|
||
authGroup.POST("/groups/:room_id/nickname", api.UpdateMyNicknameHandler) // 修改我在群里的昵称（群名片）
|
||
authGroup.GET("/group-notifications", api.GetGroupNotificationsHandler)
|
||
authGroup.GET("/calls/history", api.GetCallHistoryHandler)
|
||
|
||
// 扫码登录（App 侧，需登录：扫码/确认/取消）
|
||
authGroup.POST("/qrcode/scan", api.ScanQRCodeHandler)
|
||
authGroup.POST("/qrcode/confirm", api.ConfirmQRCodeHandler)
|
||
authGroup.POST("/qrcode/cancel", api.CancelQRCodeHandler)
|
||
|
||
// AI 机器人：配置读写/测试/机器人增删改仅管理员(id=1)，列表所有登录用户可见
|
||
authGroup.GET("/ai/config", api.GetAIConfigHandler)
|
||
authGroup.POST("/ai/config", api.SaveAIConfigHandler)
|
||
authGroup.POST("/ai/config/test", api.TestAIConfigHandler)
|
||
authGroup.GET("/ai/bots", api.ListAIBotsHandler)
|
||
authGroup.POST("/ai/bots", api.CreateAIBotHandler)
|
||
authGroup.POST("/ai/bots/update/:id", api.UpdateAIBotHandler)
|
||
authGroup.POST("/ai/bots/delete/:id", api.DeleteAIBotHandler)
|
||
|
||
// 附件管理
|
||
authGroup.POST("/attachments/upload", api.UploadAttachmentHandler)
|
||
authGroup.GET("/attachments", api.GetAttachmentsHandler)
|
||
authGroup.GET("/attachments/:id", api.GetAttachmentHandler)
|
||
authGroup.POST("/attachments/delete/:id", api.DeleteAttachmentHandler)
|
||
|
||
// 朋友圈管理
|
||
api.RegisterMomentRoutes(authGroup)
|
||
|
||
// 通话管理（音视频）
|
||
api.RegisterCallRoutes(authGroup)
|
||
|
||
// 媒体服务器管理
|
||
api.RegisterMediaRoutes(authGroup)
|
||
}
|
||
}
|
||
|
||
// 步骤10: 启动HTTP服务器（支持优雅关闭）
|
||
// 原实现 r.Run 阻塞至进程被信号强杀，defer 的清理逻辑（StopWorkerPool 等）不会执行，
|
||
// MediaServer（SFU/RTMP/FFmpeg 子进程）、协程池、Redis/DB 连接都得不到有序释放，
|
||
// 可能残留端口占用与孤儿 FFmpeg 进程。
|
||
// 改为 http.Server + 信号监听：收到 SIGINT/SIGTERM 后先停止接收新请求并限时等待存量请求，
|
||
// 再依次关闭各组件，最后退出。
|
||
port := viper.GetString("app.port") // 从配置文件读取端口号
|
||
srv := &http.Server{
|
||
Addr: ":" + port,
|
||
Handler: r,
|
||
}
|
||
|
||
// HTTP 服务放到独立协程启动，主协程留下来等退出信号
|
||
go func() {
|
||
log.Printf("🚀 服务启动在端口: %s", port)
|
||
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||
log.Fatalf("❌ HTTP 服务启动失败: %v", err)
|
||
}
|
||
}()
|
||
|
||
// 阻塞等待 SIGINT(Ctrl+C) / SIGTERM(kill) 退出信号
|
||
quit := make(chan os.Signal, 1)
|
||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
||
<-quit
|
||
log.Println("🛑 收到退出信号，开始优雅关闭...")
|
||
|
||
// 限时 10 秒等待存量 HTTP 请求处理完毕（超时则强制返回）
|
||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
if err := srv.Shutdown(ctx); err != nil {
|
||
log.Printf("⚠️ HTTP 服务关闭异常: %v", err)
|
||
}
|
||
|
||
// 有序释放各组件资源：
|
||
// 1. 媒体服务器（关闭所有房间、SFU、RTMP 监听、FFmpeg 子进程）
|
||
mediaserver.GetServer().Stop()
|
||
// 2. WebSocket 协程池（defer 中也有一次，ants 的 Release 可重复调用，安全）
|
||
ws.StopWorkerPool()
|
||
// 3. Redis 连接
|
||
if err := rdb.Close(); err != nil {
|
||
log.Printf("⚠️ Redis 关闭异常: %v", err)
|
||
}
|
||
// 4. MySQL 连接池
|
||
if sqlDB, err := db.DB(); err == nil {
|
||
if err := sqlDB.Close(); err != nil {
|
||
log.Printf("⚠️ MySQL 关闭异常: %v", err)
|
||
}
|
||
}
|
||
log.Println("✅ 服务已优雅退出")
|
||
}
|