【系统架构设计百科】Slack 架构：实时协作的工程挑战

# Slack 早期架构组件（2013-2015）
web_tier:
  language: PHP 5.x
  server: Apache + mod_php
  framework: 自研 MVC 框架
  session: Memcached

data_tier:
  primary_db: MySQL 5.6（单主多从）
  cache: Memcached
  search: Solr
  queue: Redis + 自研 Job Runner

realtime_tier:
  protocol: WebSocket
  server: Node.js（独立进程）
  pubsub: Redis Pub/Sub

package gateway

import (
    "context"
    "net/http"
    "sync"
    "time"

    "github.com/gorilla/websocket"
)

// ConnRegistry 管理所有活跃的 WebSocket 连接
type ConnRegistry struct {
    mu    sync.RWMutex
    conns map[string]*ClientConn // key: connectionID
    users map[string][]*ClientConn // key: userID
}

// ClientConn 表示一个客户端连接
type ClientConn struct {
    ID        string
    UserID    string
    TeamID    string
    Conn      *websocket.Conn
    Channels  []string
    SendCh    chan []byte
    CreatedAt time.Time
    ctx       context.Context
    cancel    context.CancelFunc
}

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

// HandleWebSocket 处理新的 WebSocket 连接请求
func (reg *ConnRegistry) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
    // 认证检查
    token := r.URL.Query().Get("token")
    userInfo, err := authenticateToken(token)
    if err != nil {
        http.Error(w, "Unauthorized", http.StatusUnauthorized)
        return
    }

    // 升级 HTTP 连接为 WebSocket
    wsConn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        return
    }

    ctx, cancel := context.WithCancel(r.Context())
    client := &ClientConn{
        ID:        generateConnID(),
        UserID:    userInfo.UserID,
        TeamID:    userInfo.TeamID,
        Conn:      wsConn,
        Channels:  userInfo.Channels,
        SendCh:    make(chan []byte, 256),
        CreatedAt: time.Now(),
        ctx:       ctx,
        cancel:    cancel,
    }

    // 注册连接
    reg.register(client)

    // 订阅用户所属频道
    for _, ch := range client.Channels {
        subscribeToPubSub(client.TeamID, ch, client)
    }

    // 启动读写 goroutine
    go client.readPump(reg)
    go client.writePump()
}

// readPump 从 WebSocket 读取客户端消息
func (c *ClientConn) readPump(reg *ConnRegistry) {
    defer func() {
        reg.unregister(c)
        c.Conn.Close()
    }()

    c.Conn.SetReadLimit(maxMessageSize)
    c.Conn.SetReadDeadline(time.Now().Add(pongWait))
    c.Conn.SetPongHandler(func(string) error {
        c.Conn.SetReadDeadline(time.Now().Add(pongWait))
        return nil
    })

    for {
        _, message, err := c.Conn.ReadMessage()
        if err != nil {
            break
        }
        processIncomingMessage(c, message)
    }
}

// writePump 向 WebSocket 写入服务端消息
func (c *ClientConn) writePump() {
    ticker := time.NewTicker(pingPeriod)
    defer func() {
        ticker.Stop()
        c.Conn.Close()
    }()

    for {
        select {
        case message, ok := <-c.SendCh:
            c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
            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(writeWait))
            if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        case <-c.ctx.Done():
            return
        }
    }
}

const (
    maxMessageSize = 65536
    pongWait       = 60 * time.Second
    pingPeriod     = 54 * time.Second
    writeWait      = 10 * time.Second
)

// ConnectionLocator 负责定位用户连接所在的网关节点
type ConnectionLocator struct {
    redisClient *redis.Client
}

// LocateUser 查找用户的所有活跃连接
func (l *ConnectionLocator) LocateUser(userID string) ([]GatewayEndpoint, error) {
    key := fmt.Sprintf("user:conns:%s", userID)
    members, err := l.redisClient.SMembers(context.Background(), key).Result()
    if err != nil {
        return nil, fmt.Errorf("failed to locate user %s: %w", userID, err)
    }

    endpoints := make([]GatewayEndpoint, 0, len(members))
    for _, member := range members {
        ep, err := parseGatewayEndpoint(member)
        if err != nil {
            continue
        }
        endpoints = append(endpoints, ep)
    }
    return endpoints, nil
}

// RegisterConnection 注册新连接到 Redis
func (l *ConnectionLocator) RegisterConnection(conn *ClientConn, gatewayAddr string) error {
    key := fmt.Sprintf("user:conns:%s", conn.UserID)
    value := fmt.Sprintf("%s|%s|%d", gatewayAddr, conn.ID, conn.CreatedAt.Unix())

    pipe := l.redisClient.Pipeline()
    pipe.SAdd(context.Background(), key, value)
    pipe.Expire(context.Background(), key, 2*time.Hour)
    _, err := pipe.Exec(context.Background())
    return err
}

// ReconnectManager 客户端重连管理
type ReconnectManager struct {
    maxRetries    int
    baseDelay     time.Duration
    maxDelay      time.Duration
    lastSeqNum    int64
}

// GetBackoffDuration 计算指数退避的等待时间
func (rm *ReconnectManager) GetBackoffDuration(attempt int) time.Duration {
    if attempt >= rm.maxRetries {
        return rm.maxDelay
    }

    delay := rm.baseDelay * time.Duration(1<<uint(attempt))
    if delay > rm.maxDelay {
        delay = rm.maxDelay
    }

    // 添加 0-25% 的随机抖动
    jitter := time.Duration(rand.Int63n(int64(delay) / 4))
    return delay + jitter
}

// HandleReconnect 处理客户端重连请求
func HandleReconnect(conn *ClientConn, lastSeqNum int64) error {
    // 查询该用户在断线期间错过的消息
    missedMessages, err := fetchMessagesSince(conn.UserID, conn.Channels, lastSeqNum)
    if err != nil {
        return fmt.Errorf("failed to fetch missed messages: %w", err)
    }

    // 按顺序补发消息
    for _, msg := range missedMessages {
        data, _ := json.Marshal(msg)
        select {
        case conn.SendCh <- data:
        default:
            return fmt.Errorf("send channel full for conn %s", conn.ID)
        }
    }
    return nil
}

// FanoutStrategy 决定消息的扇出策略
type FanoutStrategy struct {
    smallChannelThreshold int
    activeWindowDuration  time.Duration
}

// DetermineFanoutPlan 根据频道规模和成员状态确定扇出计划
func (fs *FanoutStrategy) DetermineFanoutPlan(
    channelID string,
    members []Member,
    onlineUsers map[string]bool,
) *FanoutPlan {
    plan := &FanoutPlan{
        ChannelID:   channelID,
        DirectPush:  make([]string, 0),
        DeferredPull: make([]string, 0),
    }

    if len(members) <= fs.smallChannelThreshold {
        // 小频道：对所有在线成员直接推送
        for _, m := range members {
            if onlineUsers[m.UserID] {
                plan.DirectPush = append(plan.DirectPush, m.UserID)
            } else {
                plan.DeferredPull = append(plan.DeferredPull, m.UserID)
            }
        }
        plan.Strategy = "fan_out_on_write"
        return plan
    }

    // 大频道：混合策略
    now := time.Now()
    for _, m := range members {
        if onlineUsers[m.UserID] && now.Sub(m.LastActiveAt) < fs.activeWindowDuration {
            plan.DirectPush = append(plan.DirectPush, m.UserID)
        } else {
            plan.DeferredPull = append(plan.DeferredPull, m.UserID)
        }
    }
    plan.Strategy = "hybrid"
    return plan
}

// FanoutPlan 扇出执行计划
type FanoutPlan struct {
    ChannelID    string
    Strategy     string
    DirectPush   []string // 立即推送的用户
    DeferredPull []string // 延迟拉取的用户
}

// BatchDispatcher 批量消息分发器
type BatchDispatcher struct {
    batchSize    int
    flushInterval time.Duration
    buffers      map[string][]Message // key: gatewayAddr
    mu           sync.Mutex
}

// Dispatch 将消息加入批量缓冲区
func (bd *BatchDispatcher) Dispatch(gatewayAddr string, msg Message) {
    bd.mu.Lock()
    defer bd.mu.Unlock()

    bd.buffers[gatewayAddr] = append(bd.buffers[gatewayAddr], msg)

    if len(bd.buffers[gatewayAddr]) >= bd.batchSize {
        bd.flush(gatewayAddr)
    }
}

// flush 将缓冲区中的消息批量发送到目标网关
func (bd *BatchDispatcher) flush(gatewayAddr string) {
    messages := bd.buffers[gatewayAddr]
    if len(messages) == 0 {
        return
    }

    bd.buffers[gatewayAddr] = make([]Message, 0, bd.batchSize)

    go func(addr string, msgs []Message) {
        batch := MessageBatch{
            GatewayAddr: addr,
            Messages:    msgs,
            Timestamp:   time.Now(),
        }
        sendBatchToGateway(batch)
    }(gatewayAddr, messages)
}

// WorkspaceSnapshot 用户工作区快照
type WorkspaceSnapshot struct {
    UserID       string                 `json:"user_id"`
    TeamID       string                 `json:"team_id"`
    Version      int64                  `json:"version"`
    GeneratedAt  time.Time              `json:"generated_at"`

    // 频道数据
    Channels     []ChannelSummary       `json:"channels"`
    DMs          []DirectMessageSummary `json:"dms"`

    // 用户数据
    Self         UserProfile            `json:"self"`
    TeamMembers  []UserBasicInfo        `json:"team_members"`

    // 未读状态
    UnreadCounts map[string]UnreadInfo  `json:"unread_counts"`

    // 偏好设置
    Preferences  UserPreferences        `json:"preferences"`
}

// ChannelSummary 频道摘要信息
type ChannelSummary struct {
    ID            string    `json:"id"`
    Name          string    `json:"name"`
    Topic         string    `json:"topic"`
    Purpose       string    `json:"purpose"`
    MemberCount   int       `json:"member_count"`
    LastMessageTS string    `json:"last_message_ts"`
    LastReadTS    string    `json:"last_read_ts"`
    IsMuted       bool      `json:"is_muted"`
    IsArchived    bool      `json:"is_archived"`
}

// UnreadInfo 未读信息
type UnreadInfo struct {
    ChannelID     string `json:"channel_id"`
    UnreadCount   int    `json:"unread_count"`
    MentionCount  int    `json:"mention_count"`
    LastReadTS    string `json:"last_read_ts"`
}

// ChangeEventProcessor 变更事件处理器
type ChangeEventProcessor struct {
    cache        *SnapshotCache
    redisClient  *redis.Client
}

// ChangeEvent 数据变更事件
type ChangeEvent struct {
    Type      string          `json:"type"`
    TeamID    string          `json:"team_id"`
    ChannelID string          `json:"channel_id,omitempty"`
    UserID    string          `json:"user_id,omitempty"`
    Payload   json.RawMessage `json:"payload"`
    Timestamp int64           `json:"ts"`
}

// StartListening 开始监听变更事件
func (p *ChangeEventProcessor) StartListening(ctx context.Context) {
    pubsub := p.redisClient.Subscribe(ctx,
        "flannel:changes:messages",
        "flannel:changes:channels",
        "flannel:changes:users",
        "flannel:changes:unreads",
    )
    defer pubsub.Close()

    ch := pubsub.Channel()
    for {
        select {
        case msg := <-ch:
            var event ChangeEvent
            if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil {
                log.Printf("failed to unmarshal change event: %v", err)
                continue
            }
            p.processEvent(event)
        case <-ctx.Done():
            return
        }
    }
}

// processEvent 处理单个变更事件
func (p *ChangeEventProcessor) processEvent(event ChangeEvent) {
    switch event.Type {
    case "new_message":
        p.cache.UpdateLatestMessage(event.TeamID, event.ChannelID, event.Payload)
    case "channel_created":
        p.cache.AddChannel(event.TeamID, event.Payload)
    case "channel_archived":
        p.cache.ArchiveChannel(event.TeamID, event.ChannelID)
    case "member_joined_channel":
        p.cache.AddChannelMember(event.TeamID, event.ChannelID, event.UserID)
    case "unread_updated":
        p.cache.UpdateUnreadCount(event.TeamID, event.UserID, event.Payload)
    case "user_profile_changed":
        p.cache.UpdateUserProfile(event.TeamID, event.UserID, event.Payload)
    }
}

// 消息服务的 gRPC 接口定义
syntax = "proto3";

package slack.messaging.v1;

service MessageService {
    // 发送消息
    rpc SendMessage(SendMessageRequest) returns (SendMessageResponse);

    // 获取频道历史消息
    rpc GetChannelHistory(GetChannelHistoryRequest) returns (GetChannelHistoryResponse);

    // 编辑消息
    rpc EditMessage(EditMessageRequest) returns (EditMessageResponse);

    // 删除消息
    rpc DeleteMessage(DeleteMessageRequest) returns (DeleteMessageResponse);

    // 搜索消息
    rpc SearchMessages(SearchMessagesRequest) returns (SearchMessagesResponse);
}

message SendMessageRequest {
    string channel_id = 1;
    string user_id = 2;
    string text = 3;
    string thread_ts = 4; // 可选：回复某条消息
    repeated Attachment attachments = 5;
    repeated Block blocks = 6;
}

message SendMessageResponse {
    bool ok = 1;
    string message_ts = 2;
    string channel_id = 3;
    MessageError error = 4;
}

message GetChannelHistoryRequest {
    string channel_id = 1;
    string latest = 2; // 时间戳上界
    string oldest = 3; // 时间戳下界
    int32 limit = 4;   // 默认 100
    bool inclusive = 5;
}

message GetChannelHistoryResponse {
    bool ok = 1;
    repeated Message messages = 2;
    bool has_more = 3;
    ResponseMetadata response_metadata = 4;
}

-- Vitess VSchema 定义（简化版）
-- 定义 messages 表的分片规则

-- 创建 vindexes（虚拟索引，用于路由查询到正确分片）
-- team_channel_hash: 基于 team_id 和 channel_id 的复合哈希

CREATE TABLE messages (
    team_id       BIGINT NOT NULL,
    channel_id    VARCHAR(32) NOT NULL,
    message_ts    VARCHAR(32) NOT NULL,
    user_id       VARCHAR(32) NOT NULL,
    text          TEXT,
    thread_ts     VARCHAR(32) DEFAULT NULL,
    edited_ts     VARCHAR(32) DEFAULT NULL,
    is_deleted    TINYINT DEFAULT 0,
    attachments   JSON DEFAULT NULL,
    reactions     JSON DEFAULT NULL,
    created_at    DATETIME NOT NULL,
    updated_at    DATETIME NOT NULL,
    PRIMARY KEY (team_id, channel_id, message_ts),
    INDEX idx_channel_ts (channel_id, message_ts),
    INDEX idx_user_ts (user_id, message_ts),
    INDEX idx_thread (channel_id, thread_ts, message_ts)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

{
    "sharded": true,
    "vindexes": {
        "team_channel_hash": {
            "type": "xxhash",
            "params": {}
        },
        "team_id_lookup": {
            "type": "consistent_lookup",
            "params": {
                "table": "team_channel_lookup",
                "from": "team_id,channel_id",
                "to": "keyspace_id"
            }
        }
    },
    "tables": {
        "messages": {
            "column_vindexes": [
                {
                    "columns": ["team_id", "channel_id"],
                    "name": "team_channel_hash"
                }
            ]
        },
        "channel_members": {
            "column_vindexes": [
                {
                    "columns": ["team_id", "channel_id"],
                    "name": "team_channel_hash"
                }
            ]
        }
    }
}

// VitessMigrationRouter 迁移期间的流量路由器
type VitessMigrationRouter struct {
    phase         MigrationPhase
    legacyDB      *sql.DB
    vitessDB      *sql.DB
    readPercent   int // Vitess 承担的读流量百分比
    featureFlags  *FeatureFlagService
}

type MigrationPhase int

const (
    PhaseShadow      MigrationPhase = iota // 阶段一：影子流量
    PhaseReadMigrate                        // 阶段二：读迁移
    PhaseWriteMigrate                       // 阶段三：写迁移
    PhaseComplete                           // 阶段四：完全切换
)

// RouteQuery 根据迁移阶段路由查询
func (r *VitessMigrationRouter) RouteQuery(
    ctx context.Context,
    query string,
    args []interface{},
    isWrite bool,
) (*sql.Rows, error) {
    switch r.phase {
    case PhaseShadow:
        // 所有操作使用旧库，同时写入 Vitess 做比对
        result, err := r.legacyDB.QueryContext(ctx, query, args...)
        if isWrite {
            go r.shadowWrite(ctx, query, args)
        }
        return result, err

    case PhaseReadMigrate:
        if isWrite {
            return r.legacyDB.QueryContext(ctx, query, args...)
        }
        // 按百分比分流读请求
        if rand.Intn(100) < r.readPercent {
            return r.vitessDB.QueryContext(ctx, query, args...)
        }
        return r.legacyDB.QueryContext(ctx, query, args...)

    case PhaseWriteMigrate:
        if isWrite {
            return r.vitessDB.QueryContext(ctx, query, args...)
        }
        return r.vitessDB.QueryContext(ctx, query, args...)

    case PhaseComplete:
        return r.vitessDB.QueryContext(ctx, query, args...)
    }

    return r.legacyDB.QueryContext(ctx, query, args...)
}

// shadowWrite 影子写入 Vitess 并比对结果
func (r *VitessMigrationRouter) shadowWrite(
    ctx context.Context,
    query string,
    args []interface{},
) {
    _, err := r.vitessDB.ExecContext(ctx, query, args...)
    if err != nil {
        metrics.IncrCounter("vitess.shadow.write.error", 1)
        log.Printf("shadow write failed: %v", err)
    }
}

// 搜索查询构建器（简化示例）
public class SlackSearchQueryBuilder {

    private final AclResolver aclResolver;
    private final QueryParser queryParser;

    public SlackSearchQueryBuilder(AclResolver aclResolver, QueryParser queryParser) {
        this.aclResolver = aclResolver;
        this.queryParser = queryParser;
    }

    /**
     * 构建带有访问控制的搜索查询
     */
    public SearchQuery buildQuery(SearchRequest request) {
        // 解析用户输入的搜索词
        ParsedQuery parsed = queryParser.parse(request.getQueryText());

        // 获取用户有权访问的频道列表
        Set<String> accessibleChannels = aclResolver.getAccessibleChannels(
            request.getUserId(),
            request.getTeamId()
        );

        // 构建带有 ACL 过滤的查询
        SearchQuery query = SearchQuery.builder()
            .teamId(request.getTeamId())
            .textQuery(parsed.getTextClause())
            .channelFilter(accessibleChannels)
            .dateRange(parsed.getDateRange())
            .fromUser(parsed.getFromUser())
            .hasAttachment(parsed.hasAttachment())
            .sortBy(request.getSortBy())
            .limit(request.getLimit())
            .offset(request.getOffset())
            .build();

        // 添加搜索结果的高亮配置
        query.setHighlightConfig(HighlightConfig.builder()
            .preTag("<mark>")
            .postTag("</mark>")
            .fragmentSize(200)
            .numberOfFragments(3)
            .build());

        return query;
    }
}

/**
 * 搜索结果排序器：综合考虑相关性、时效性和用户互动
 */
public class SlackSearchRanker {

    private static final double RECENCY_WEIGHT = 0.3;
    private static final double RELEVANCE_WEIGHT = 0.5;
    private static final double INTERACTION_WEIGHT = 0.2;

    public double computeScore(SearchHit hit, SearchContext context) {
        double relevanceScore = hit.getTextRelevanceScore();

        // 时效性得分：越近的消息得分越高
        long ageHours = Duration.between(
            hit.getTimestamp(), Instant.now()
        ).toHours();
        double recencyScore = 1.0 / (1.0 + Math.log1p(ageHours));

        // 互动得分：基于用户与该频道和发送者的互动频率
        double interactionScore = computeInteractionScore(
            context.getUserId(),
            hit.getChannelId(),
            hit.getSenderId()
        );

        return RELEVANCE_WEIGHT * relevanceScore
             + RECENCY_WEIGHT * recencyScore
             + INTERACTION_WEIGHT * interactionScore;
    }

    private double computeInteractionScore(
        String userId, String channelId, String senderId
    ) {
        int channelActivity = getRecentChannelActivity(userId, channelId);
        int senderInteraction = getRecentUserInteraction(userId, senderId);
        return Math.min(1.0, (channelActivity + senderInteraction) / 100.0);
    }
}

// RateLimiter 多维度限流器
type RateLimiter struct {
    teamLimits    map[string]*TokenBucket
    userLimits    map[string]*TokenBucket
    globalLimit   *TokenBucket
    mu            sync.RWMutex
}

// TokenBucket 令牌桶实现
type TokenBucket struct {
    tokens     float64
    capacity   float64
    refillRate float64 // 每秒补充的令牌数
    lastRefill time.Time
    mu         sync.Mutex
}

// Allow 检查是否允许请求通过
func (tb *TokenBucket) Allow() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    now := time.Now()
    elapsed := now.Sub(tb.lastRefill).Seconds()
    tb.tokens = math.Min(tb.capacity, tb.tokens+elapsed*tb.refillRate)
    tb.lastRefill = now

    if tb.tokens >= 1 {
        tb.tokens--
        return true
    }
    return false
}

// CheckRateLimit 执行多层限流检查
func (rl *RateLimiter) CheckRateLimit(teamID, userID string) error {
    // 全局限流
    if !rl.globalLimit.Allow() {
        return &RateLimitError{
            Level:   "global",
            Message: "Global rate limit exceeded",
            RetryAfter: 5 * time.Second,
        }
    }

    // 团队级限流
    rl.mu.RLock()
    teamBucket, exists := rl.teamLimits[teamID]
    rl.mu.RUnlock()
    if exists && !teamBucket.Allow() {
        return &RateLimitError{
            Level:   "team",
            Message: "Team rate limit exceeded",
            RetryAfter: 2 * time.Second,
        }
    }

    // 用户级限流
    rl.mu.RLock()
    userBucket, exists := rl.userLimits[userID]
    rl.mu.RUnlock()
    if exists && !userBucket.Allow() {
        return &RateLimitError{
            Level:   "user",
            Message: "User rate limit exceeded",
            RetryAfter: 1 * time.Second,
        }
    }

    return nil
}

// CircuitBreaker 断路器实现
type CircuitBreaker struct {
    name          string
    state         CircuitState
    failCount     int64
    successCount  int64
    threshold     int64       // 触发断路的连续失败次数
    resetTimeout  time.Duration
    lastFailure   time.Time
    mu            sync.Mutex
}

type CircuitState int

const (
    StateClosed   CircuitState = iota // 正常状态
    StateOpen                         // 断开状态
    StateHalfOpen                     // 半开状态（试探）
)

// Execute 通过断路器执行操作
func (cb *CircuitBreaker) Execute(fn func() error) error {
    cb.mu.Lock()
    state := cb.state

    switch state {
    case StateOpen:
        // 检查是否到了重试时间
        if time.Since(cb.lastFailure) > cb.resetTimeout {
            cb.state = StateHalfOpen
            cb.mu.Unlock()
            return cb.tryHalfOpen(fn)
        }
        cb.mu.Unlock()
        return &CircuitOpenError{
            Service:    cb.name,
            RetryAfter: cb.resetTimeout - time.Since(cb.lastFailure),
        }

    case StateHalfOpen:
        cb.mu.Unlock()
        return cb.tryHalfOpen(fn)

    default: // StateClosed
        cb.mu.Unlock()
        return cb.tryClosed(fn)
    }
}

func (cb *CircuitBreaker) tryClosed(fn func() error) error {
    err := fn()
    cb.mu.Lock()
    defer cb.mu.Unlock()

    if err != nil {
        cb.failCount++
        cb.lastFailure = time.Now()
        if cb.failCount >= cb.threshold {
            cb.state = StateOpen
            log.Printf("circuit breaker [%s] opened after %d failures", cb.name, cb.failCount)
        }
        return err
    }

    cb.failCount = 0
    return nil
}

func (cb *CircuitBreaker) tryHalfOpen(fn func() error) error {
    err := fn()
    cb.mu.Lock()
    defer cb.mu.Unlock()

    if err != nil {
        cb.state = StateOpen
        cb.lastFailure = time.Now()
        return err
    }

    cb.state = StateClosed
    cb.failCount = 0
    cb.successCount++
    log.Printf("circuit breaker [%s] closed after successful probe", cb.name)
    return nil
}

// 统一的请求追踪中间件
type TracingMiddleware struct {
    tracer  opentracing.Tracer
    service string
}

func (tm *TracingMiddleware) Wrap(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        spanCtx, _ := tm.tracer.Extract(
            opentracing.HTTPHeaders,
            opentracing.HTTPHeadersCarrier(r.Header),
        )

        span := tm.tracer.StartSpan(
            r.URL.Path,
            ext.RPCServerOption(spanCtx),
        )
        defer span.Finish()

        ext.HTTPMethod.Set(span, r.Method)
        ext.HTTPUrl.Set(span, r.URL.String())
        span.SetTag("service", tm.service)
        span.SetTag("team_id", r.Header.Get("X-Slack-Team-ID"))

        ctx := opentracing.ContextWithSpan(r.Context(), span)
        rw := &responseWriter{ResponseWriter: w, statusCode: 200}

        next.ServeHTTP(rw, r.WithContext(ctx))

        ext.HTTPStatusCode.Set(span, uint16(rw.statusCode))
        if rw.statusCode >= 500 {
            ext.Error.Set(span, true)
        }
    })
}

// PaginatedMemberLoader 分页成员加载器
type PaginatedMemberLoader struct {
    vitessClient *VitessClient
    cache        *MemberCache
}

type MemberPage struct {
    Members    []MemberInfo `json:"members"`
    TotalCount int          `json:"total_count"`
    NextCursor string       `json:"next_cursor"`
    HasMore    bool         `json:"has_more"`
}

// LoadMemberPage 加载一页成员数据
func (ml *PaginatedMemberLoader) LoadMemberPage(
    ctx context.Context,
    teamID string,
    channelID string,
    cursor string,
    pageSize int,
) (*MemberPage, error) {
    // 优先从缓存获取
    cacheKey := fmt.Sprintf("members:%s:%s:%s:%d", teamID, channelID, cursor, pageSize)
    if cached, ok := ml.cache.Get(cacheKey); ok {
        return cached.(*MemberPage), nil
    }

    // 使用游标分页查询 Vitess
    query := `
        SELECT cm.user_id, u.display_name, u.avatar_hash, cm.role, cm.joined_at
        FROM channel_members cm
        JOIN users u ON cm.user_id = u.user_id AND cm.team_id = u.team_id
        WHERE cm.team_id = ? AND cm.channel_id = ?
    `
    args := []interface{}{teamID, channelID}

    if cursor != "" {
        query += " AND cm.user_id > ?"
        args = append(args, cursor)
    }

    query += " ORDER BY cm.user_id LIMIT ?"
    args = append(args, pageSize+1) // 多取一条以判断是否还有更多

    rows, err := ml.vitessClient.Query(ctx, query, args...)
    if err != nil {
        return nil, fmt.Errorf("failed to query members: %w", err)
    }
    defer rows.Close()

    members := make([]MemberInfo, 0, pageSize)
    for rows.Next() {
        var m MemberInfo
        if err := rows.Scan(&m.UserID, &m.DisplayName, &m.AvatarHash, &m.Role, &m.JoinedAt); err != nil {
            return nil, err
        }
        members = append(members, m)
    }

    hasMore := len(members) > pageSize
    if hasMore {
        members = members[:pageSize]
    }

    page := &MemberPage{
        Members:    members,
        TotalCount: ml.getCachedTotalCount(teamID, channelID),
        HasMore:    hasMore,
    }
    if len(members) > 0 {
        page.NextCursor = members[len(members)-1].UserID
    }

    ml.cache.Set(cacheKey, page, 30*time.Second)
    return page, nil
}

// PresenceIndex 在线状态预索引
type PresenceIndex struct {
    redisClient *redis.Client
}

// GetOnlineMembers 获取指定频道的在线成员
func (pi *PresenceIndex) GetOnlineMembers(
    ctx context.Context,
    teamID string,
    channelID string,
) ([]string, error) {
    // 使用 Redis 集合交集：频道成员集合 AND 在线用户集合
    channelKey := fmt.Sprintf("channel:members:%s:%s", teamID, channelID)
    onlineKey := fmt.Sprintf("team:online:%s", teamID)

    // SINTERSTORE 将交集结果存入临时集合
    resultKey := fmt.Sprintf("tmp:online_in_channel:%s:%s:%d",
        teamID, channelID, time.Now().UnixNano())

    pipe := pi.redisClient.Pipeline()
    pipe.SInterStore(ctx, resultKey, channelKey, onlineKey)
    pipe.Expire(ctx, resultKey, 10*time.Second)
    pipe.SMembers(ctx, resultKey)
    results, err := pipe.Exec(ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to compute online members: %w", err)
    }

    // 获取交集结果
    members := results[2].(*redis.StringSliceCmd).Val()
    // 清理临时 key
    go pi.redisClient.Del(context.Background(), resultKey)

    return members, nil
}