11 - 直播系统设计

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "sync"
    "time"
    
    "github.com/google/uuid"
    "github.com/gorilla/mux"
    "github.com/gorilla/websocket"
)

// ==================== 数据结构 ====================

// LiveRoom 直播间
type LiveRoom struct {
    RoomID         string    `json:"room_id"`
    Title          string    `json:"title"`
    StreamerUserID int64     `json:"streamer_user_id"`
    Status         string    `json:"status"` // created | living | ended
    PushURL        string    `json:"push_url"`
    PlayURLs       *PlayURLs `json:"play_urls"`
    OnlineCount    int       `json:"online_count"`
    TotalViews     int       `json:"total_views"`
    StartedAt      time.Time `json:"started_at,omitempty"`
    CreatedAt      time.Time `json:"created_at"`
}

// PlayURLs 播放地址
type PlayURLs struct {
    RTMP string `json:"rtmp"`
    FLV  string `json:"flv"`
    HLS  string `json:"hls"`
}

// ChatMessage 聊天消息
type ChatMessage struct {
    MessageID   string                 `json:"message_id"`
    RoomID      string                 `json:"room_id"`
    UserID      int64                  `json:"user_id"`
    Nickname    string                 `json:"nickname"`
    MessageType string                 `json:"message_type"` // text | gift | system
    Content     string                 `json:"content"`
    Metadata    map[string]interface{} `json:"metadata,omitempty"`
    Timestamp   int64                  `json:"timestamp"`
}

// ==================== 直播间服务 ====================

// LiveRoomService 直播间服务
type LiveRoomService struct {
    rooms map[string]*LiveRoom
    mu    sync.RWMutex
}

// NewLiveRoomService 创建直播间服务
func NewLiveRoomService() *LiveRoomService {
    return &LiveRoomService{
        rooms: make(map[string]*LiveRoom),
    }
}

// CreateRoom 创建直播间
func (s *LiveRoomService) CreateRoom(title string, streamerUserID int64) (*LiveRoom, error) {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    roomID := generateRoomID()
    
    room := &LiveRoom{
        RoomID:         roomID,
        Title:          title,
        StreamerUserID: streamerUserID,
        Status:         "created",
        PushURL:        fmt.Sprintf("rtmp://push.example.com/live/%s", roomID),
        PlayURLs: &PlayURLs{
            RTMP: fmt.Sprintf("rtmp://play.example.com/live/%s", roomID),
            FLV:  fmt.Sprintf("https://play.example.com/live/%s.flv", roomID),
            HLS:  fmt.Sprintf("https://play.example.com/live/%s.m3u8", roomID),
        },
        OnlineCount: 0,
        TotalViews:  0,
        CreatedAt:   time.Now(),
    }
    
    s.rooms[roomID] = room
    
    return room, nil
}

// StartLive 开始直播
func (s *LiveRoomService) StartLive(roomID string) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    room, exists := s.rooms[roomID]
    if !exists {
        return fmt.Errorf("直播间不存在")
    }
    
    if room.Status == "living" {
        return fmt.Errorf("直播已开始")
    }
    
    room.Status = "living"
    room.StartedAt = time.Now()
    
    return nil
}

// GetRoom 获取直播间信息
func (s *LiveRoomService) GetRoom(roomID string) (*LiveRoom, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    
    room, exists := s.rooms[roomID]
    if !exists {
        return nil, fmt.Errorf("直播间不存在")
    }
    
    return room, nil
}

// IncrOnlineCount 增加在线人数
func (s *LiveRoomService) IncrOnlineCount(roomID string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    if room, exists := s.rooms[roomID]; exists {
        room.OnlineCount++
        room.TotalViews++
    }
}

// DecrOnlineCount 减少在线人数
func (s *LiveRoomService) DecrOnlineCount(roomID string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    if room, exists := s.rooms[roomID]; exists && room.OnlineCount > 0 {
        room.OnlineCount--
    }
}

// ==================== 聊天室服务 ====================

var upgrader = websocket.Upgrader{
    CheckOrigin: func(r *http.Request) bool {
        return true
    },
}

// ChatRoomService 聊天室服务
type ChatRoomService struct {
    // 房间 -> 客户端连接
    rooms map[string]map[*Client]bool
    
    // 广播消息通道
    broadcast chan *ChatMessage
    
    // 注册通道
    register chan *Client
    
    // 注销通道
    unregister chan *Client
    
    mu sync.RWMutex
}

// Client WebSocket 客户端
type Client struct {
    RoomID   string
    UserID   int64
    Nickname string
    Conn     *websocket.Conn
    Send     chan []byte
}

// NewChatRoomService 创建聊天室服务
func NewChatRoomService() *ChatRoomService {
    return &ChatRoomService{
        rooms:      make(map[string]map[*Client]bool),
        broadcast:  make(chan *ChatMessage, 256),
        register:   make(chan *Client),
        unregister: make(chan *Client),
    }
}

// Run 运行聊天室服务
func (s *ChatRoomService) Run() {
    for {
        select {
        case client := <-s.register:
            s.mu.Lock()
            if s.rooms[client.RoomID] == nil {
                s.rooms[client.RoomID] = make(map[*Client]bool)
            }
            s.rooms[client.RoomID][client] = true
            s.mu.Unlock()
            
            fmt.Printf("用户 %d 加入直播间 %s\n", client.UserID, client.RoomID)
            
        case client := <-s.unregister:
            s.mu.Lock()
            if clients, ok := s.rooms[client.RoomID]; ok {
                if _, exists := clients[client]; exists {
                    delete(clients, client)
                    close(client.Send)
                    
                    if len(clients) == 0 {
                        delete(s.rooms, client.RoomID)
                    }
                }
            }
            s.mu.Unlock()
            
            fmt.Printf("用户 %d 离开直播间 %s\n", client.UserID, client.RoomID)
            
        case message := <-s.broadcast:
            s.mu.RLock()
            clients := s.rooms[message.RoomID]
            s.mu.RUnlock()
            
            // 广播消息
            data, _ := json.Marshal(message)
            for client := range clients {
                select {
                case client.Send <- data:
                default:
                    close(client.Send)
                    delete(clients, client)
                }
            }
        }
    }
}

// Broadcast 广播消息
func (s *ChatRoomService) Broadcast(message *ChatMessage) {
    s.broadcast <- message
}

// HandleWebSocket 处理 WebSocket 连接
func (s *ChatRoomService) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
    // 升级为 WebSocket
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        fmt.Printf("WebSocket upgrade failed: %v\n", err)
        return
    }
    
    // 从 URL 参数获取信息
    roomID := r.URL.Query().Get("room_id")
    userID := int64(12345) // 实际应该从 token 解析
    nickname := "用户" + uuid.New().String()[:8]
    
    client := &Client{
        RoomID:   roomID,
        UserID:   userID,
        Nickname: nickname,
        Conn:     conn,
        Send:     make(chan []byte, 256),
    }
    
    // 注册客户端
    s.register <- client
    
    // 启动读写协程
    go client.writePump()
    go client.readPump(s)
}

// readPump 读取客户端消息
func (c *Client) readPump(service *ChatRoomService) {
    defer func() {
        service.unregister <- c
        c.Conn.Close()
    }()
    
    c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
    c.Conn.SetPongHandler(func(string) error {
        c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
        return nil
    })
    
    for {
        _, messageData, err := c.Conn.ReadMessage()
        if err != nil {
            break
        }
        
        // 解析消息
        var msg map[string]interface{}
        if err := json.Unmarshal(messageData, &msg); err != nil {
            continue
        }
        
        // 构建聊天消息
        chatMsg := &ChatMessage{
            MessageID:   generateMessageID(),
            RoomID:      c.RoomID,
            UserID:      c.UserID,
            Nickname:    c.Nickname,
            MessageType: "text",
            Content:     msg["content"].(string),
            Timestamp:   time.Now().Unix(),
        }
        
        // 广播
        service.Broadcast(chatMsg)
    }
}

// writePump 发送消息给客户端
func (c *Client) writePump() {
    ticker := time.NewTicker(54 * time.Second)
    defer func() {
        ticker.Stop()
        c.Conn.Close()
    }()
    
    for {
        select {
        case message, ok := <-c.Send:
            c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
            if !ok {
                c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }
            
            if err := c.Conn.WriteMessage(websocket.TextMessage, message); err != nil {
                return
            }
            
        case <-ticker.C:
            c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
            if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        }
    }
}

// ==================== HTTP Handlers ====================

type LiveHandler struct {
    liveService *LiveRoomService
    chatService *ChatRoomService
}

func (h *LiveHandler) CreateRoomHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        Title          string `json:"title"`
        StreamerUserID int64  `json:"streamer_user_id"`
    }
    
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    room, err := h.liveService.CreateRoom(req.Title, req.StreamerUserID)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
        "data":    room,
    })
}

func (h *LiveHandler) GetRoomHandler(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    roomID := vars["room_id"]
    
    room, err := h.liveService.GetRoom(roomID)
    if err != nil {
        http.Error(w, err.Error(), http.StatusNotFound)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
        "data":    room,
    })
}

func (h *LiveHandler) StartLiveHandler(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    roomID := vars["room_id"]
    
    if err := h.liveService.StartLive(roomID); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
    })
}

// ==================== 辅助函数 ====================

func generateRoomID() string {
    return uuid.New().String()[:8]
}

func generateMessageID() string {
    return "msg_" + uuid.New().String()
}

// ==================== Main ====================

func main() {
    liveService := NewLiveRoomService()
    chatService := NewChatRoomService()
    
    // 启动聊天室服务
    go chatService.Run()
    
    handler := &LiveHandler{
        liveService: liveService,
        chatService: chatService,
    }
    
    // 设置路由
    r := mux.NewRouter()
    r.HandleFunc("/api/v1/live/rooms", handler.CreateRoomHandler).Methods("POST")
    r.HandleFunc("/api/v1/live/rooms/{room_id}", handler.GetRoomHandler).Methods("GET")
    r.HandleFunc("/api/v1/live/rooms/{room_id}/start", handler.StartLiveHandler).Methods("POST")
    r.HandleFunc("/ws/chat", chatService.HandleWebSocket)
    
    fmt.Println("直播服务启动在 :8080")
    http.ListenAndServe(":8080", r)
}

// ==================== 分布式聊天室（Redis Pub/Sub）====================

package distributed

import (
    "context"
    "encoding/json"
    "fmt"
    
    "github.com/go-redis/redis/v8"
)

// DistributedChatService 分布式聊天服务
type DistributedChatService struct {
    redis *redis.Client
    
    // 本地连接管理
    localClients map[string]map[*Client]bool
    
    // 订阅的房间
    subscriptions map[string]*redis.PubSub
}

// NewDistributedChatService 创建分布式聊天服务
func NewDistributedChatService(redisClient *redis.Client) *DistributedChatService {
    return &DistributedChatService{
        redis:         redisClient,
        localClients:  make(map[string]map[*Client]bool),
        subscriptions: make(map[string]*redis.PubSub),
    }
}

// SubscribeRoom 订阅房间消息
func (s *DistributedChatService) SubscribeRoom(roomID string) {
    if _, exists := s.subscriptions[roomID]; exists {
        return
    }
    
    ctx := context.Background()
    pubsub := s.redis.Subscribe(ctx, fmt.Sprintf("chat:room:%s", roomID))
    s.subscriptions[roomID] = pubsub
    
    // 接收消息
    go func() {
        ch := pubsub.Channel()
        for msg := range ch {
            var chatMsg ChatMessage
            if err := json.Unmarshal([]byte(msg.Payload), &chatMsg); err != nil {
                continue
            }
            
            // 广播给本地客户端
            s.broadcastToLocal(roomID, &chatMsg)
        }
    }()
}

// PublishMessage 发布消息到 Redis
func (s *DistributedChatService) PublishMessage(message *ChatMessage) error {
    ctx := context.Background()
    
    // 1. 发布到 Redis Pub/Sub
    data, _ := json.Marshal(message)
    channelName := fmt.Sprintf("chat:room:%s", message.RoomID)
    
    if err := s.redis.Publish(ctx, channelName, data).Err(); err != nil {
        return err
    }
    
    // 2. 异步写入 Kafka（持久化）
    go s.writeToKafka(message)
    
    return nil
}

// broadcastToLocal 广播给本地客户端
func (s *DistributedChatService) broadcastToLocal(roomID string, message *ChatMessage) {
    clients := s.localClients[roomID]
    data, _ := json.Marshal(message)
    
    for client := range clients {
        select {
        case client.Send <- data:
        default:
            close(client.Send)
            delete(clients, client)
        }
    }
}

// writeToKafka 写入 Kafka
func (s *DistributedChatService) writeToKafka(message *ChatMessage) {
    // 实际实现：发送到 Kafka
    // kafkaProducer.Produce("chat-messages", message)
}

// ==================== 弹幕限流 ====================

import (
    "time"
    
    "golang.org/x/time/rate"
)

// DanmakuLimiter 弹幕限流器
type DanmakuLimiter struct {
    // 用户级限流
    userLimiters map[int64]*rate.Limiter
    
    // 房间级限流
    roomLimiters map[string]*rate.Limiter
}

// NewDanmakuLimiter 创建弹幕限流器
func NewDanmakuLimiter() *DanmakuLimiter {
    return &DanmakuLimiter{
        userLimiters: make(map[int64]*rate.Limiter),
        roomLimiters: make(map[string]*rate.Limiter),
    }
}

// AllowUser 检查用户是否允许发送弹幕
func (l *DanmakuLimiter) AllowUser(userID int64) bool {
    limiter, exists := l.userLimiters[userID]
    if !exists {
        // 每用户每秒 3 条
        limiter = rate.NewLimiter(rate.Every(time.Second/3), 3)
        l.userLimiters[userID] = limiter
    }
    
    return limiter.Allow()
}

// AllowRoom 检查房间是否允许接收弹幕
func (l *DanmakuLimiter) AllowRoom(roomID string) bool {
    limiter, exists := l.roomLimiters[roomID]
    if !exists {
        // 每房间每秒 100 条
        limiter = rate.NewLimiter(rate.Every(time.Second/100), 100)
        l.roomLimiters[roomID] = limiter
    }
    
    return limiter.Allow()
}

// ==================== 礼物系统 ====================

package gift

import (
    "context"
    "fmt"
)

// GiftService 礼物服务
type GiftService struct {
    db    *sql.DB
    redis *redis.Client
}

// SendGift 赠送礼物
func (s *GiftService) SendGift(
    roomID string,
    senderUserID, receiverUserID int64,
    giftID int,
    count int,
) (*GiftOrder, error) {
    // 1. 查询礼物信息
    gift, err := s.getGift(giftID)
    if err != nil {
        return nil, err
    }
    
    // 2. 计算总金额
    totalAmount := gift.Price * float64(count)
    
    // 3. 扣除用户余额（调用支付服务）
    if err := s.deductBalance(senderUserID, totalAmount); err != nil {
        return nil, err
    }
    
    // 4. 创建礼物订单
    order := &GiftOrder{
        OrderID:         generateOrderID(),
        RoomID:          roomID,
        SenderUserID:    senderUserID,
        ReceiverUserID:  receiverUserID,
        GiftID:          giftID,
        Count:           count,
        UnitPrice:       gift.Price,
        TotalAmount:     totalAmount,
        Status:          "success",
        Timestamp:       time.Now().Unix(),
    }
    
    _, err = s.db.Exec(`
        INSERT INTO gift_orders (
            order_id, room_id, sender_user_id, receiver_user_id,
            gift_id, count, unit_price, total_amount, status, timestamp
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    `, order.OrderID, order.RoomID, order.SenderUserID, order.ReceiverUserID,
        order.GiftID, order.Count, order.UnitPrice, order.TotalAmount,
        order.Status, order.Timestamp)
    
    if err != nil {
        // 回滚余额
        s.refundBalance(senderUserID, totalAmount)
        return nil, err
    }
    
    // 5. 发送礼物消息到聊天室
    giftMessage := &ChatMessage{
        MessageID:   generateMessageID(),
        RoomID:      roomID,
        UserID:      senderUserID,
        MessageType: "gift",
        Content:     fmt.Sprintf("赠送了 %d 个 %s", count, gift.Name),
        Metadata: map[string]interface{}{
            "gift_id":       giftID,
            "gift_name":     gift.Name,
            "count":         count,
            "animation_url": gift.AnimationURL,
        },
        Timestamp: time.Now().Unix(),
    }
    
    s.chatService.PublishMessage(giftMessage)
    
    // 6. 更新主播收益（异步）
    go s.updateStreamerIncome(receiverUserID, totalAmount*0.5) // 50% 分成
    
    return order, nil
}

type GiftOrder struct {
    OrderID        string
    RoomID         string
    SenderUserID   int64
    ReceiverUserID int64
    GiftID         int
    Count          int
    UnitPrice      float64
    TotalAmount    float64
    Status         string
    Timestamp      int64
}

type Gift struct {
    GiftID       int
    Name         string
    Price        float64
    AnimationURL string
}

func (s *GiftService) getGift(giftID int) (*Gift, error) {
    var gift Gift
    err := s.db.QueryRow(`
        SELECT gift_id, name, price, animation_url 
        FROM gifts 
        WHERE gift_id = ?
    `, giftID).Scan(&gift.GiftID, &gift.Name, &gift.Price, &gift.AnimationURL)
    
    return &gift, err
}

func (s *GiftService) deductBalance(userID int64, amount float64) error {
    // 调用支付服务
    return nil
}

func (s *GiftService) refundBalance(userID int64, amount float64) error {
    // 退款
    return nil
}

func (s *GiftService) updateStreamerIncome(userID int64, amount float64) error {
    // 更新主播收益
    return nil
}

func generateOrderID() string {
    return "gift_" + uuid.New().String()
}

// ==================== CDN 调度 ====================

package cdn

import (
    "fmt"
    "math"
)

// CDNScheduler CDN 调度器
type CDNScheduler struct {
    nodes []*CDNNode
}

// CDNNode CDN 节点
type CDNNode struct {
    NodeID    string
    Region    string  // "华北" | "华东" | "华南"
    Bandwidth float64 // 可用带宽（Gbps）
    Load      float64 // 负载（0-1）
    Latency   int     // 延迟（ms）
}

// SelectBestNode 选择最佳 CDN 节点
func (s *CDNScheduler) SelectBestNode(userRegion string, userIP string) *CDNNode {
    var bestNode *CDNNode
    bestScore := -1.0
    
    for _, node := range s.nodes {
        // 计算综合分数
        score := s.calculateScore(node, userRegion)
        
        if score > bestScore {
            bestScore = score
            bestNode = node
        }
    }
    
    return bestNode
}

// calculateScore 计算节点分数
func (s *CDNScheduler) calculateScore(node *CDNNode, userRegion string) float64 {
    // 1. 地域匹配（同地域加分）
    regionScore := 0.0
    if node.Region == userRegion {
        regionScore = 1.0
    } else {
        regionScore = 0.5
    }
    
    // 2. 负载（负载越低越好）
    loadScore := 1.0 - node.Load
    
    // 3. 延迟（延迟越低越好）
    latencyScore := 1.0 / (1.0 + float64(node.Latency)/100.0)
    
    // 4. 带宽（带宽越大越好）
    bandwidthScore := math.Log(node.Bandwidth + 1)
    
    // 加权综合
    score := regionScore*0.4 + loadScore*0.3 + latencyScore*0.2 + bandwidthScore*0.1
    
    return score
}

// ==================== 多清晰度转码 ====================

package transcode

// TranscodeService 转码服务
type TranscodeService struct {
    ffmpegPath string
}

// TranscodeProfile 转码配置
type TranscodeProfile struct {
    Name       string // "超清" | "高清" | "标清"
    Resolution string // "1920x1080" | "1280x720" | "640x480"
    Bitrate    string // "2000k" | "1000k" | "500k"
    Framerate  int    // 30 | 25 | 15
}

// Transcode 转码
func (s *TranscodeService) Transcode(inputURL string, profiles []*TranscodeProfile) error {
    for _, profile := range profiles {
        outputURL := fmt.Sprintf("%s_%s.flv", inputURL, profile.Name)
        
        // 构建 ffmpeg 命令
        cmd := fmt.Sprintf(`
            %s -i %s \
            -c:v libx264 -preset veryfast \
            -b:v %s -r %d \
            -s %s \
            -c:a aac -b:a 128k \
            -f flv %s
        `, s.ffmpegPath, inputURL, profile.Bitrate, profile.Framerate,
            profile.Resolution, outputURL)
        
        // 异步执行
        go s.executeFFmpeg(cmd)
    }
    
    return nil
}

func (s *TranscodeService) executeFFmpeg(cmd string) {
    // 执行 ffmpeg 命令
    // exec.Command("sh", "-c", cmd).Run()
}

// ==================== 直播回放 ====================

package replay

import (
    "fmt"
    "time"
)

// ReplayService 回放服务
type ReplayService struct {
    db         *sql.DB
    storageURL string
}

// CreateReplay 创建回放
func (s *ReplayService) CreateReplay(roomID string, startTime, endTime time.Time) (*Replay, error) {
    replayID := generateReplayID()
    
    // 1. 从录制文件生成回放
    recordFile := fmt.Sprintf("%s/record/%s_%d.flv",
        s.storageURL, roomID, startTime.Unix())
    
    replayFile := fmt.Sprintf("%s/replay/%s.mp4", s.storageURL, replayID)
    
    // 2. 转换为 MP4 格式
    if err := s.convertToMP4(recordFile, replayFile); err != nil {
        return nil, err
    }
    
    // 3. 生成封面图
    coverURL := s.generateCover(replayFile)
    
    // 4. 保存到数据库
    duration := endTime.Sub(startTime).Seconds()
    
    replay := &Replay{
        ReplayID:  replayID,
        RoomID:    roomID,
        Title:     "直播回放",
        VideoURL:  replayFile,
        CoverURL:  coverURL,
        Duration:  int(duration),
        StartTime: startTime,
        EndTime:   endTime,
        CreatedAt: time.Now(),
    }
    
    _, err := s.db.Exec(`
        INSERT INTO replays (
            replay_id, room_id, title, video_url, cover_url,
            duration, start_time, end_time, created_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    `, replay.ReplayID, replay.RoomID, replay.Title, replay.VideoURL,
        replay.CoverURL, replay.Duration, replay.StartTime, replay.EndTime,
        replay.CreatedAt)
    
    return replay, err
}

type Replay struct {
    ReplayID  string
    RoomID    string
    Title     string
    VideoURL  string
    CoverURL  string
    Duration  int
    StartTime time.Time
    EndTime   time.Time
    CreatedAt time.Time
}

func (s *ReplayService) convertToMP4(inputFile, outputFile string) error {
    // ffmpeg 转换
    return nil
}

func (s *ReplayService) generateCover(videoFile string) string {
    // 生成封面图
    return ""
}

func generateReplayID() string {
    return "replay_" + uuid.New().String()
}

// 动态码率调整
type AdaptiveBitrateService struct {
    profiles []*BitrateProfile
}

type BitrateProfile struct {
    Bitrate    int    // kbps
    Resolution string
    Quality    string // "超清" | "高清" | "标清"
}

// SelectProfile 根据网络状况选择码率
func (s *AdaptiveBitrateService) SelectProfile(bandwidth int) *BitrateProfile {
    // 选择最接近但不超过带宽的配置
    for i := len(s.profiles) - 1; i >= 0; i-- {
        if s.profiles[i].Bitrate <= bandwidth*80/100 { // 预留 20% 缓冲
            return s.profiles[i]
        }
    }
    
    // 返回最低画质
    return s.profiles[0]
}

// 长连接优化
type ConnectionPool struct {
    maxConns    int
    activeConns int
    mu          sync.Mutex
}

// Acquire 获取连接
func (p *ConnectionPool) Acquire() bool {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if p.activeConns >= p.maxConns {
        return false
    }
    
    p.activeConns++
    return true
}

// Release 释放连接
func (p *ConnectionPool) Release() {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    p.activeConns--
}

var (
    liveRoomCount = prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "live_room_count",
            Help: "Current live room count",
        },
    )
    
    onlineUserCount = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "live_online_user_count",
            Help: "Online user count per room",
        },
        []string{"room_id"},
    )
    
    messageLatency = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "live_message_latency_seconds",
            Help:    "Message latency",
            Buckets: []float64{0.001, 0.01, 0.1, 0.5, 1},
        },
        []string{"message_type"},
    )
    
    streamBitrate = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "live_stream_bitrate_kbps",
            Help: "Stream bitrate",
        },
        []string{"room_id", "quality"},
    )
)

// 发送消息
redis.Publish("chat:room:12345", message)

// 订阅消息
pubsub := redis.Subscribe("chat:room:12345")
for msg := range pubsub.Channel() {
    broadcastToLocalClients(msg)
}

// 用户级：每秒 3 条
userLimiter := rate.NewLimiter(3, 3)

// 房间级：每秒 100 条
roomLimiter := rate.NewLimiter(100, 100)

func SendGift(userID, streamerID int64, giftID int) error {
    tx, _ := db.Begin()
    defer tx.Rollback()
    
    // 1. 扣款
    tx.Exec(`UPDATE accounts SET balance = balance - ? WHERE user_id = ?`, giftPrice, userID)
    
    // 2. 创建订单
    tx.Exec(`INSERT INTO gift_orders ...`)
    
    // 3. 增加主播收益
    tx.Exec(`UPDATE accounts SET balance = balance + ? WHERE user_id = ?`, giftPrice*0.5, streamerID)
    
    return tx.Commit()
}

// 推流服务器采集
type StreamMetrics struct {
    RoomID    string
    FPS       int
    Bitrate   int // kbps
    PacketLoss float64
}

// 上报到 Prometheus
streamBitrate.WithLabelValues(roomID, "hd").Set(float64(bitrate))