03 - IM 即时通讯系统设计

// 消息类型
const (
    MsgTypeText     = "text"     // 文本
    MsgTypeImage    = "image"    // 图片
    MsgTypeVoice    = "voice"    // 语音
    MsgTypeVideo    = "video"    // 视频
    MsgTypeFile     = "file"     // 文件
    MsgTypeLocation = "location" // 位置
    MsgTypeCard     = "card"     // 名片
    MsgTypeSystem   = "system"   // 系统消息
)

// 协议类型
const (
    ProtocolMessage = "message"  // 普通消息
    ProtocolAck     = "ack"      // 确认消息
    ProtocolPing    = "ping"     // 心跳请求
    ProtocolPong    = "pong"     // 心跳响应
    ProtocolKickoff = "kickoff"  // 踢下线
    ProtocolSync    = "sync"     // 同步请求
)

package server

import (
    "github.com/gorilla/websocket"
    "sync"
)

// Connection WebSocket 连接封装
type Connection struct {
    UserID     int64
    DeviceID   string
    Conn       *websocket.Conn
    SendChan   chan []byte
    CloseChan  chan struct{}
    LastActive time.Time
}

// ConnectionManager 连接管理器
type ConnectionManager struct {
    connections map[int64]*Connection  // user_id -> connection
    mu          sync.RWMutex
}

func NewConnectionManager() *ConnectionManager {
    return &ConnectionManager{
        connections: make(map[int64]*Connection),
    }
}

// AddConnection 添加连接
func (cm *ConnectionManager) AddConnection(userID int64, conn *Connection) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
    cm.connections[userID] = conn

    log.Infof("用户 %d 已连接，当前在线: %d", userID, len(cm.connections))
}

// RemoveConnection 移除连接
func (cm *ConnectionManager) RemoveConnection(userID int64) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
    delete(cm.connections, userID)

    log.Infof("用户 %d 已断开，当前在线: %d", userID, len(cm.connections))
}

// GetConnection 获取连接
func (cm *ConnectionManager) GetConnection(userID int64) (*Connection, bool) {
    cm.mu.RLock()
    defer cm.mu.RUnlock()
    conn, ok := cm.connections[userID]
    return conn, ok
}

// SendToUser 发送消息给用户
func (cm *ConnectionManager) SendToUser(userID int64, data []byte) error {
    conn, ok := cm.GetConnection(userID)
    if !ok {
        return errors.New("用户不在线")
    }

    select {
    case conn.SendChan <- data:
        return nil
    case <-time.After(5 * time.Second):
        return errors.New("发送超时")
    }
}

package handler

import (
    "encoding/json"
    "time"
)

type MessageHandler struct {
    connMgr *ConnectionManager
    db      *sql.DB
}

// HandleMessage 处理收到的消息
func (h *MessageHandler) HandleMessage(conn *Connection, data []byte) error {
    var msg Message
    if err := json.Unmarshal(data, &msg); err != nil {
        return err
    }

    switch msg.Type {
    case ProtocolMessage:
        return h.handleChatMessage(conn, &msg)
    case ProtocolAck:
        return h.handleAck(conn, &msg)
    case ProtocolPing:
        return h.handlePing(conn)
    default:
        return errors.New("未知消息类型")
    }
}

// handleChatMessage 处理聊天消息
func (h *MessageHandler) handleChatMessage(conn *Connection, msg *Message) error {
    // 1. 生成消息 ID
    msg.MsgID = generateMsgID()
    msg.Timestamp = time.Now().UnixMilli()
    msg.FromUID = conn.UserID

    // 2. 持久化到数据库
    if err := h.saveMessage(msg); err != nil {
        return err
    }

    // 3. 发送给接收方
    if msg.ToUID != 0 {
        // 单聊
        return h.sendToUser(msg)
    } else if msg.ToGID != 0 {
        // 群聊
        return h.sendToGroup(msg)
    }

    return nil
}

// sendToUser 发送单聊消息
func (h *MessageHandler) sendToUser(msg *Message) error {
    // 检查接收方是否在线
    if h.connMgr.IsOnline(msg.ToUID) {
        // 在线，直接推送
        data, _ := json.Marshal(msg)
        return h.connMgr.SendToUser(msg.ToUID, data)
    } else {
        // 离线，存储离线消息
        return h.saveOfflineMessage(msg.ToUID, msg.MsgID)
    }
}

// sendToGroup 发送群聊消息
func (h *MessageHandler) sendToGroup(msg *Message) error {
    // 查询群成员
    members, err := h.getGroupMembers(msg.ToGID)
    if err != nil {
        return err
    }

    data, _ := json.Marshal(msg)

    // 遍历群成员，发送消息
    for _, memberID := range members {
        if memberID == msg.FromUID {
            continue  // 跳过发送者自己
        }

        if h.connMgr.IsOnline(memberID) {
            h.connMgr.SendToUser(memberID, data)
        } else {
            h.saveOfflineMessage(memberID, msg.MsgID)
        }
    }

    return nil
}

// saveMessage 保存消息到数据库
func (h *MessageHandler) saveMessage(msg *Message) error {
    sql := `
        INSERT INTO message (msg_id, session_id, from_uid, to_uid, to_gid, msg_type, content, seq, timestamp)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    `
    _, err := h.db.Exec(sql, msg.MsgID, msg.SessionID, msg.FromUID, msg.ToUID, msg.ToGID,
        msg.MsgType, msg.Content, msg.Seq, msg.Timestamp)
    return err
}

// saveOfflineMessage 保存离线消息索引
func (h *MessageHandler) saveOfflineMessage(userID int64, msgID string) error {
    sql := "INSERT INTO offline_message (user_id, msg_id) VALUES (?, ?)"
    _, err := h.db.Exec(sql, userID, msgID)
    return err
}

// handlePing 处理心跳
func (h *MessageHandler) handlePing(conn *Connection) error {
    // 更新最后活跃时间
    conn.LastActive = time.Now()

    // 发送 pong 响应
    pong := Message{Type: ProtocolPong, Timestamp: time.Now().UnixMilli()}
    data, _ := json.Marshal(pong)

    return conn.Conn.WriteMessage(websocket.TextMessage, data)
}

// HeartbeatChecker 心跳检查器
func (cm *ConnectionManager) HeartbeatChecker() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        cm.mu.Lock()
        for userID, conn := range cm.connections {
            // 如果 2 分钟没有活跃，关闭连接
            if time.Since(conn.LastActive) > 2*time.Minute {
                log.Warnf("用户 %d 心跳超时，关闭连接", userID)
                conn.Conn.Close()
                delete(cm.connections, userID)
            }
        }
        cm.mu.Unlock()
    }
}

package router

import (
    "github.com/go-redis/redis/v8"
)

type UserRouter struct {
    rdb *redis.Client
}

// RegisterUser 用户连接时注册路由
func (r *UserRouter) RegisterUser(ctx context.Context, userID int64, gatewayID, connID string) error {
    key := fmt.Sprintf("user:online:%d", userID)
    value := fmt.Sprintf("%s_%s", gatewayID, connID)

    // 设置 5 分钟过期（心跳刷新）
    return r.rdb.Set(ctx, key, value, 5*time.Minute).Err()
}

// GetUserGateway 查询用户所在网关
func (r *UserRouter) GetUserGateway(ctx context.Context, userID int64) (string, string, error) {
    key := fmt.Sprintf("user:online:%d", userID)
    value, err := r.rdb.Get(ctx, key).Result()
    if err != nil {
        return "", "", err
    }

    parts := strings.Split(value, "_")
    if len(parts) != 2 {
        return "", "", errors.New("invalid format")
    }

    return parts[0], parts[1], nil  // gatewayID, connID
}

// UnregisterUser 用户断开连接时注销路由
func (r *UserRouter) UnregisterUser(ctx context.Context, userID int64) error {
    key := fmt.Sprintf("user:online:%d", userID)
    return r.rdb.Del(ctx, key).Err()
}

package delivery

import (
    "github.com/Shopify/sarama"
)

type MessageDelivery struct {
    producer sarama.SyncProducer
    router   *UserRouter
    rpcPool  *GatewayRPCPool
}

// DeliverMessage 投递消息给用户
func (d *MessageDelivery) DeliverMessage(ctx context.Context, msg *Message) error {
    // 1. 查询用户所在网关
    gatewayID, connID, err := d.router.GetUserGateway(ctx, msg.ToUID)
    if err != nil {
        // 用户不在线，保存离线消息
        return d.saveOfflineMessage(msg)
    }

    // 2. 判断是否在本网关
    if gatewayID == currentGatewayID {
        // 本地投递
        return d.deliverLocal(connID, msg)
    } else {
        // 跨网关投递（通过 Kafka）
        return d.deliverRemote(gatewayID, connID, msg)
    }
}

// deliverRemote 跨网关投递
func (d *MessageDelivery) deliverRemote(gatewayID, connID string, msg *Message) error {
    // 构造投递消息
    deliveryMsg := DeliveryMessage{
        GatewayID: gatewayID,
        ConnID:    connID,
        Message:   msg,
    }

    data, _ := json.Marshal(deliveryMsg)

    // 发送到 Kafka（topic 按 gateway_id 分区）
    partition := hashGatewayID(gatewayID) % numPartitions
    _, _, err := d.producer.SendMessage(&sarama.ProducerMessage{
        Topic:     "im-delivery",
        Key:       sarama.StringEncoder(gatewayID),
        Value:     sarama.ByteEncoder(data),
        Partition: int32(partition),
    })

    return err
}

// Kafka 消费者（每个网关订阅自己的分区）
type DeliveryConsumer struct {
    connMgr *ConnectionManager
}

func (c *DeliveryConsumer) Consume(msg *sarama.ConsumerMessage) error {
    var deliveryMsg DeliveryMessage
    if err := json.Unmarshal(msg.Value, &deliveryMsg); err != nil {
        return err
    }

    // 本地投递
    return c.connMgr.SendToConnection(deliveryMsg.ConnID, deliveryMsg.Message)
}

package ack

type AckManager struct {
    rdb *redis.Client
}

// SendMessage 发送消息并等待 ACK
func (h *MessageHandler) SendMessage(msg *Message) error {
    // 1. 发送消息
    data, _ := json.Marshal(msg)
    if err := h.delivery.DeliverMessage(context.Background(), msg); err != nil {
        return err
    }

    // 2. 等待 ACK（超时 10 秒）
    ackKey := fmt.Sprintf("msg:ack:%s", msg.MsgID)
    for i := 0; i < 20; i++ {  // 重试 20 次，每次 500ms
        exists, _ := h.rdb.Exists(context.Background(), ackKey).Result()
        if exists == 1 {
            // 收到 ACK
            return nil
        }
        time.Sleep(500 * time.Millisecond)
    }

    // 超时未收到 ACK，重试或保存离线消息
    log.Warnf("消息 %s 未收到 ACK", msg.MsgID)
    return errors.New("ACK timeout")
}

// HandleAck 处理客户端 ACK
func (h *MessageHandler) HandleAck(conn *Connection, ack *AckMessage) error {
    ackKey := fmt.Sprintf("msg:ack:%s", ack.MsgID)
    return h.rdb.Set(context.Background(), ackKey, "1", 10*time.Second).Err()
}

// PullOfflineMessages 拉取离线消息
func (h *MessageHandler) PullOfflineMessages(userID int64, lastSeq int64) ([]*Message, error) {
    // 1. 查询离线消息索引
    query := `
        SELECT msg_id FROM offline_message
        WHERE user_id = ?
        ORDER BY created_at ASC
        LIMIT 100
    `
    rows, err := h.db.Query(query, userID)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var msgIDs []string
    for rows.Next() {
        var msgID string
        rows.Scan(&msgID)
        msgIDs = append(msgIDs, msgID)
    }

    // 2. 批量查询消息内容
    if len(msgIDs) == 0 {
        return []*Message{}, nil
    }

    placeholders := strings.Repeat("?,", len(msgIDs))
    placeholders = placeholders[:len(placeholders)-1]

    query = fmt.Sprintf(`
        SELECT msg_id, from_uid, to_uid, to_gid, msg_type, content, seq, timestamp
        FROM message
        WHERE msg_id IN (%s)
        ORDER BY timestamp ASC
    `, placeholders)

    args := make([]interface{}, len(msgIDs))
    for i, id := range msgIDs {
        args[i] = id
    }

    rows, err = h.db.Query(query, args...)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var messages []*Message
    for rows.Next() {
        var msg Message
        rows.Scan(&msg.MsgID, &msg.FromUID, &msg.ToUID, &msg.ToGID, &msg.MsgType,
            &msg.Content, &msg.Seq, &msg.Timestamp)
        messages = append(messages, &msg)
    }

    // 3. 删除已拉取的离线消息索引
    deleteSQL := fmt.Sprintf("DELETE FROM offline_message WHERE user_id = ? AND msg_id IN (%s)", placeholders)
    h.db.Exec(deleteSQL, append([]interface{}{userID}, args...)...)

    return messages, nil
}

package sync

type MultiDeviceSync struct {
    router *UserRouter
}

// BroadcastToAllDevices 广播消息到用户所有设备
func (s *MultiDeviceSync) BroadcastToAllDevices(ctx context.Context, userID int64, msg *Message) error {
    // 1. 查询用户所有在线设备
    key := fmt.Sprintf("user:sessions:%d", userID)
    sessions, err := s.rdb.HGetAll(ctx, key).Result()
    if err != nil {
        return err
    }

    // 2. 遍历所有设备，发送消息
    for deviceID, gatewayConn := range sessions {
        parts := strings.Split(gatewayConn, "_")
        gatewayID, connID := parts[0], parts[1]

        log.Infof("发送消息到用户 %d 设备 %s (网关: %s)", userID, deviceID, gatewayID)

        if err := s.delivery.DeliverToGateway(gatewayID, connID, msg); err != nil {
            log.Errorf("发送失败: %v", err)
        }
    }

    return nil
}

package seq

type SequenceManager struct {
    rdb *redis.Client
}

// AllocateSeq 为消息分配序号（会话内单调递增）
func (s *SequenceManager) AllocateSeq(ctx context.Context, sessionID string) (int64, error) {
    key := fmt.Sprintf("session:seq:%s", sessionID)
    seq, err := s.rdb.Incr(ctx, key).Result()
    return seq, err
}

// 消息按 seq 排序
type MessageSorter []*Message

func (m MessageSorter) Len() int           { return len(m) }
func (m MessageSorter) Less(i, j int) bool { return m[i].Seq < m[j].Seq }
func (m MessageSorter) Swap(i, j int)      { m[i], m[j] = m[j], m[i] }

// 客户端按 seq 去重和排序
func (c *Client) ProcessMessages(messages []*Message) {
    // 1. 去重（检查 seq 是否已处理）
    var newMessages []*Message
    for _, msg := range messages {
        if !c.isSeqProcessed(msg.Seq) {
            newMessages = append(newMessages, msg)
            c.markSeqProcessed(msg.Seq)
        }
    }

    // 2. 排序
    sort.Sort(MessageSorter(newMessages))

    // 3. 处理消息
    for _, msg := range newMessages {
        c.handleMessage(msg)
    }
}

// 写扩散：普通群
func (h *MessageHandler) SendGroupMessage_WriteFanout(msg *Message) error {
    // 1. 保存一份群消息
    if err := h.saveMessage(msg); err != nil {
        return err
    }

    // 2. 查询群成员
    members, _ := h.getGroupMembers(msg.ToGID)

    // 3. 为每个成员创建一条消息记录（写扩散）
    for _, memberID := range members {
        if memberID == msg.FromUID {
            continue
        }

        // 创建会话消息
        sessionMsg := *msg
        sessionMsg.SessionID = fmt.Sprintf("group_%d_user_%d", msg.ToGID, memberID)
        sessionMsg.Seq, _ = h.seqMgr.AllocateSeq(context.Background(), sessionMsg.SessionID)

        // 保存到用户会话表
        h.saveSessionMessage(memberID, &sessionMsg)

        // 推送或离线
        if h.router.IsOnline(memberID) {
            h.delivery.DeliverMessage(context.Background(), &sessionMsg)
        } else {
            h.saveOfflineMessage(memberID, sessionMsg.MsgID)
        }
    }

    return nil
}

// 读扩散：超级群
func (h *MessageHandler) SendGroupMessage_ReadFanout(msg *Message) error {
    // 1. 只保存一份群消息
    if err := h.saveMessage(msg); err != nil {
        return err
    }

    // 2. 推送给在线成员（通过 Kafka 广播）
    h.kafka.Publish("group-message", GroupMessageEvent{
        GroupID: msg.ToGID,
        Message: msg,
    })

    return nil
}

// 读扩散：拉取群消息
func (h *MessageHandler) PullGroupMessages(userID, groupID, lastSeq int64) ([]*Message, error) {
    // 查询群消息（读时过滤）
    query := `
        SELECT msg_id, from_uid, msg_type, content, seq, timestamp
        FROM message
        WHERE to_gid = ? AND seq > ?
        ORDER BY seq ASC
        LIMIT 100
    `
    // ...
}

package sharding

// 消息表分片：按 session_id 哈希 + 按月分表
func GetMessageTable(sessionID string, timestamp int64) string {
    // 1. 按 session_id 哈希，分 16 个库
    dbIndex := crc32.ChecksumIEEE([]byte(sessionID)) % 16

    // 2. 按时间分表（每月一张表）
    month := time.Unix(timestamp/1000, 0).Format("200601")  // 202511

    return fmt.Sprintf("db%d.message_%s", dbIndex, month)
}

// 会话表分片：按 user_id 分 256 张表
func GetSessionTable(userID int64) string {
    dbIndex := userID % 16
    tableIndex := (userID / 16) % 256
    return fmt.Sprintf("db%d.session_%d", dbIndex, tableIndex)
}

// 使用示例
func (h *MessageHandler) SaveMessage(msg *Message) error {
    tableName := GetMessageTable(msg.SessionID, msg.Timestamp)

    sql := fmt.Sprintf(`
        INSERT INTO %s (msg_id, session_id, from_uid, to_uid, msg_type, content, seq, timestamp)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    `, tableName)

    _, err := h.db.Exec(sql, msg.MsgID, msg.SessionID, msg.FromUID, msg.ToUID,
        msg.MsgType, msg.Content, msg.Seq, msg.Timestamp)
    return err
}

package reconnect

// ClientReconnect 客户端断线重连
func (c *Client) Reconnect() error {
    // 1. 指数退避重连
    backoff := 1 * time.Second
    maxBackoff := 60 * time.Second

    for {
        if err := c.Connect(); err == nil {
            // 连接成功
            c.onReconnected()
            return nil
        }

        log.Infof("重连失败，%v 后重试", backoff)
        time.Sleep(backoff)

        backoff *= 2
        if backoff > maxBackoff {
            backoff = maxBackoff
        }
    }
}

// onReconnected 重连成功后的处理
func (c *Client) onReconnected() {
    // 1. 拉取离线消息
    lastSeq := c.getLastSeq()
    messages, _ := c.pullOfflineMessages(lastSeq)

    // 2. 处理消息
    c.ProcessMessages(messages)

    // 3. 同步未读数
    c.syncUnreadCount()

    log.Info("重连成功，消息已同步")
}

package idgen

import (
    "fmt"
    "sync"
    "time"
)

// SnowflakeIDGenerator 雪花算法 ID 生成器
type SnowflakeIDGenerator struct {
    mu           sync.Mutex
    timestamp    int64
    machineID    int64  // 机器 ID (10 bits)
    sequence     int64  // 序列号 (12 bits)
}

const (
    epoch           = int64(1609459200000)  // 2021-01-01
    machineBits     = uint(10)
    sequenceBits    = uint(12)
    machineMax      = int64(-1) ^ (int64(-1) << machineBits)
    sequenceMask    = int64(-1) ^ (int64(-1) << sequenceBits)
    machineShift    = sequenceBits
    timestampShift  = machineBits + sequenceBits
)

func NewSnowflakeIDGenerator(machineID int64) *SnowflakeIDGenerator {
    return &SnowflakeIDGenerator{
        machineID: machineID,
    }
}

func (s *SnowflakeIDGenerator) NextID() (int64, error) {
    s.mu.Lock()
    defer s.mu.Unlock()

    now := time.Now().UnixNano() / 1e6

    if now < s.timestamp {
        return 0, fmt.Errorf("clock moved backwards")
    }

    if now == s.timestamp {
        s.sequence = (s.sequence + 1) & sequenceMask
        if s.sequence == 0 {
            for now <= s.timestamp {
                now = time.Now().UnixNano() / 1e6
            }
        }
    } else {
        s.sequence = 0
    }

    s.timestamp = now

    id := ((now - epoch) << timestampShift) |
          (s.machineID << machineShift) |
          s.sequence

    return id, nil
}

func GenerateMessageID(machineID int64) string {
    gen := NewSnowflakeIDGenerator(machineID)
    id, _ := gen.NextID()
    return fmt.Sprintf("%d", id)
}

package consistent

import (
    "hash/crc32"
    "sort"
    "sync"
)

type ConsistentHash struct {
    ring       map[uint32]string  // 哈希环
    sortedKeys []uint32           // 排序的哈希值
    nodes      map[string]bool    // 节点集合
    replicas   int                // 虚拟节点数
    mu         sync.RWMutex
}

func NewConsistentHash(replicas int) *ConsistentHash {
    return &ConsistentHash{
        ring:     make(map[uint32]string),
        nodes:    make(map[string]bool),
        replicas: replicas,
    }
}

// AddNode 添加节点
func (c *ConsistentHash) AddNode(node string) {
    c.mu.Lock()
    defer c.mu.Unlock()

    if c.nodes[node] {
        return
    }

    c.nodes[node] = true

    // 添加虚拟节点
    for i := 0; i < c.replicas; i++ {
        hash := c.hashKey(fmt.Sprintf("%s#%d", node, i))
        c.ring[hash] = node
        c.sortedKeys = append(c.sortedKeys, hash)
    }

    sort.Slice(c.sortedKeys, func(i, j int) bool {
        return c.sortedKeys[i] < c.sortedKeys[j]
    })
}

// RemoveNode 移除节点
func (c *ConsistentHash) RemoveNode(node string) {
    c.mu.Lock()
    defer c.mu.Unlock()

    if !c.nodes[node] {
        return
    }

    delete(c.nodes, node)

    for i := 0; i < c.replicas; i++ {
        hash := c.hashKey(fmt.Sprintf("%s#%d", node, i))
        delete(c.ring, hash)
    }

    // 重建 sortedKeys
    c.sortedKeys = c.sortedKeys[:0]
    for k := range c.ring {
        c.sortedKeys = append(c.sortedKeys, k)
    }
    sort.Slice(c.sortedKeys, func(i, j int) bool {
        return c.sortedKeys[i] < c.sortedKeys[j]
    })
}

// GetNode 获取节点
func (c *ConsistentHash) GetNode(key string) string {
    c.mu.RLock()
    defer c.mu.RUnlock()

    if len(c.sortedKeys) == 0 {
        return ""
    }

    hash := c.hashKey(key)

    // 二分查找
    idx := sort.Search(len(c.sortedKeys), func(i int) bool {
        return c.sortedKeys[i] >= hash
    })

    if idx == len(c.sortedKeys) {
        idx = 0
    }

    return c.ring[c.sortedKeys[idx]]
}

func (c *ConsistentHash) hashKey(key string) uint32 {
    return crc32.ChecksumIEEE([]byte(key))
}

// 使用示例：根据 user_id 选择网关
func SelectGateway(userID int64, gateways []string) string {
    ch := NewConsistentHash(150)  // 150 个虚拟节点

    for _, gateway := range gateways {
        ch.AddNode(gateway)
    }

    return ch.GetNode(fmt.Sprintf("%d", userID))
}

package dedup

import (
    "github.com/bits-and-blooms/bloom/v3"
)

type MessageDeduplicator struct {
    bloom *bloom.BloomFilter
    rdb   *redis.Client
}

func NewMessageDeduplicator(expectedItems uint) *MessageDeduplicator {
    return &MessageDeduplicator{
        bloom: bloom.NewWithEstimates(expectedItems, 0.01),  // 1% 误判率
    }
}

// IsDuplicate 检查消息是否重复
func (d *MessageDeduplicator) IsDuplicate(ctx context.Context, clientMsgID string) (bool, error) {
    // 1. 先用布隆过滤器快速判断
    if !d.bloom.TestString(clientMsgID) {
        // 肯定不存在
        d.bloom.AddString(clientMsgID)
        return false, nil
    }

    // 2. 布隆过滤器说可能存在，进一步检查 Redis
    key := fmt.Sprintf("msg:dedup:%s", clientMsgID)
    exists, err := d.rdb.Exists(ctx, key).Result()
    if err != nil {
        return false, err
    }

    if exists == 1 {
        // 确实重复
        return true, nil
    }

    // 不重复，记录到 Redis（24 小时过期）
    d.rdb.Set(ctx, key, "1", 24*time.Hour)
    return false, nil
}

package ratelimit

type SlidingWindowRateLimiter struct {
    rdb *redis.Client
}

// Allow 判断是否允许通过（滑动窗口）
func (r *SlidingWindowRateLimiter) Allow(ctx context.Context, userID int64, limit int) (bool, error) {
    key := fmt.Sprintf("ratelimit:user:%d", userID)
    now := time.Now().UnixMilli()
    windowStart := now - 60000  // 1 分钟窗口

    // Lua 脚本保证原子性
    script := `
        local key = KEYS[1]
        local now = tonumber(ARGV[1])
        local window_start = tonumber(ARGV[2])
        local limit = tonumber(ARGV[3])

        -- 删除窗口外的记录
        redis.call('ZREMRANGEBYSCORE', key, 0, window_start)

        -- 统计窗口内的请求数
        local count = redis.call('ZCARD', key)

        if count < limit then
            -- 允许通过，记录本次请求
            redis.call('ZADD', key, now, now)
            redis.call('EXPIRE', key, 60)
            return 1
        else
            return 0
        end
    `

    result, err := r.rdb.Eval(ctx, script, []string{key}, now, windowStart, limit).Int()
    return result == 1, err
}

package compression

import (
    "bytes"
    "compress/gzip"
    "io"
)

// CompressMessage 压缩消息
func CompressMessage(data []byte) ([]byte, error) {
    var buf bytes.Buffer
    writer := gzip.NewWriter(&buf)

    if _, err := writer.Write(data); err != nil {
        return nil, err
    }

    if err := writer.Close(); err != nil {
        return nil, err
    }

    return buf.Bytes(), nil
}

// DecompressMessage 解压消息
func DecompressMessage(data []byte) ([]byte, error) {
    reader, err := gzip.NewReader(bytes.NewReader(data))
    if err != nil {
        return nil, err
    }
    defer reader.Close()

    return io.ReadAll(reader)
}

// 使用场景：消息内容 > 1KB 时压缩
func (h *MessageHandler) SendMessage(msg *Message) error {
    data, _ := json.Marshal(msg)

    if len(data) > 1024 {
        // 压缩
        compressed, _ := CompressMessage(data)
        msg.Compressed = true
        msg.Content = base64.StdEncoding.EncodeToString(compressed)
    }

    // 发送...
}

package archive

type MessageArchiver struct {
    db  *sql.DB
    oss *OSSClient
}

// ArchiveColdData 归档冷数据
func (a *MessageArchiver) ArchiveColdData() error {
    // 1. 查询 12 个月前的消息
    threshold := time.Now().AddDate(0, -12, 0)

    query := `
        SELECT * FROM message
        WHERE created_at < ?
        LIMIT 10000
    `
    rows, _ := a.db.Query(query, threshold)
    defer rows.Close()

    var messages []*Message
    for rows.Next() {
        var msg Message
        // scan...
        messages = append(messages, &msg)
    }

    if len(messages) == 0 {
        return nil
    }

    // 2. 压缩并上传到 OSS
    data, _ := json.Marshal(messages)
    compressed, _ := gzip.Compress(data)

    fileName := fmt.Sprintf("message_archive_%s.gz", time.Now().Format("20060102"))
    if err := a.oss.Upload(fileName, compressed); err != nil {
        return err
    }

    // 3. 删除原数据
    msgIDs := make([]string, len(messages))
    for i, msg := range messages {
        msgIDs[i] = msg.MsgID
    }

    placeholders := strings.Repeat("?,", len(msgIDs))
    placeholders = placeholders[:len(placeholders)-1]

    deleteSQL := fmt.Sprintf("DELETE FROM message WHERE msg_id IN (%s)", placeholders)
    args := make([]interface{}, len(msgIDs))
    for i, id := range msgIDs {
        args[i] = id
    }

    _, err := a.db.Exec(deleteSQL, args...)
    return err
}