[主持人操作] → [API 网关] → [权限状态服务] → [消息总线] → [推送网关集群] → [长连接客户端] → [本地状态机更新 UI]
                    ↑              ↑                    ↑
              [鉴权/限流]    [状态机/持久化]      [路由/分片/降级]

stateDiagram-v2
    [*] --> Attendee: 入会默认
    Attendee --> Speaker: 主持人邀请/举手通过
    Speaker --> Attendee: 主持人降级/自己下麦
    Attendee --> Host: 主持人移交
    Host --> Attendee: 移交给他人/离会
    Attendee --> Muted: 全员静音/单人静音
    Muted --> Attendee: 解除静音
    Speaker --> Muted: 全员静音覆盖
    Muted --> Speaker: 解除静音恢复发言席

message PermissionChangeCmd {
  string meeting_id = 1;
  string target_uid = 2;
  PermissionAction action = 3;   // GRANT_SPEAKER, REVOKE_SPEAKER, MUTE, UNMUTE, TRANSFER_HOST...
  string operator_uid = 4;
  int64  timestamp_ms = 5;
  uint64 version = 6;            // 乐观锁版本号
  map<string, string> ext = 7;   // 扩展字段：分组ID、优先级等
}

message PermissionSnapshot {
  string meeting_id = 1;
  string uid = 2;
  RoleState main_role = 3;       // ATTENDEE, SPEAKER, HOST
  repeated SubState sub_states = 4; // MUTED, VIDEO_OFF...
  uint64 version = 5;
  int64  updated_at_ms = 6;
}

// 客户端重连伪代码
func (c *Client) Reconnect() error {
    // 1. 建立新连接，携带 last_known_version
    conn, err := c.dialWithVersion(c.meetingID, c.uid, c.lastVersion)
    if err != nil { return err }

    // 2. 服务端返回增量变更或全量快照
    snapshot, err := c.pullSnapshotSince(conn, c.lastVersion)
    if err != nil { return err }

    // 3. 本地状态机回放，合并 UI
    c.stateMachine.Replay(snapshot.Events)
    c.lastVersion = snapshot.Version
    return nil
}

-- 权限变更事件表（事件溯源）
CREATE TABLE perm_events (
    id BIGSERIAL PRIMARY KEY,
    meeting_id VARCHAR(64) NOT NULL,
    target_uid VARCHAR(64) NOT NULL,
    action SMALLINT NOT NULL,
    operator_uid VARCHAR(64) NOT NULL,
    version BIGINT NOT NULL,          -- 全局单调递增版本号
    idempotency_key VARCHAR(128) UNIQUE, -- 客户端生成 UUID，防重复提交
    ext JSONB,
    created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX idx_meeting_version ON perm_events(meeting_id, version);

// usePermissionMachine.ts
const usePermissionMachine = (meetingId: string, uid: string) => {
  const [state, setState] = useState<PermissionState>(initialState);
  const wsRef = useRef<WebSocket>();

  useEffect(() => {
    wsRef.current = connect(meetingId, uid, state.version);
    wsRef.current.onmessage = (evt) => {
      const cmd = decode(evt.data);
      // 版本号过滤
      if (cmd.version <= state.version) return;
      // 驱动状态机
      const next = permissionReducer(state, cmd);
      setState(next);
      // 版本号持久化到 IndexedDB，刷新恢复
      idb.set('perm_version', next.version);
    };
    return () => wsRef.current?.close();
  }, []);

  // 组件仅消费 state，无感知网络细节
  return state;
};

// 连接状态机：CONNECTING → CONNECTED → RECONNECTING → POLLING
if (ws.readyState !== WebSocket.OPEN && retryCount > 3) {
  switchToPolling(5000); // 5s 轮询 /meeting/{id}/permission/snapshot
}

// 推送前合规过滤器 - 编译期强制注入
func ComplianceFilter(cmd *PermissionChangeCmd, targetUID string) *PermissionChangeCmd {
    // 1. 最小化字段：仅保留目标端渲染必需字段
    filtered := &PermissionChangeCmd{
        MeetingId:   cmd.MeetingId,
        TargetUid:   targetUID, // 强制覆盖为当前接收者，防止遍历泄露
        Action:      cmd.Action,
        Version:     cmd.Version,
        TimestampMs: cmd.TimestampMs,
    }
    // 2. 操作人脱敏：非管理员视角隐藏具体操作人
    if !isAdminView(targetUID) {
        filtered.OperatorUid = "host" // 或 "system"
    }
    // 3. 扩展字段白名单
    filtered.Ext = pickAllowList(cmd.Ext, allowExtKeys...)
    return filtered
}

graph LR
    Client[客户端 SDK] -->|能力协商| Gateway[边缘网关<br/>Envoy/NGINX + QUIC Patch]
    Gateway -->|WT 连接| WT_Handler[WT 处理器<br/>多流调度]
    Gateway -->|WS 连接| WS_Handler[WS 处理器<br/>兼容旧版]
    WT_Handler -->|Stream 0: 权限指令<br/>Stream 1: 信令<br/>Stream N: 数据| Logic[业务逻辑层<br/>无感知协议差异]
    WS_Handler --> Logic

# KEDA ScaledObject 示例：基于「每实例连接数」+「消息积压」双指标扩缩容
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: push-gateway-scaler
spec:
  scaleTargetRef:
    name: push-gateway
  pollingInterval: 15
  cooldownPeriod: 300
  minReplicaCount: 3  # 兜底高可用
  maxReplicaCount: 500
  triggers:
  - type: prometheus
    metadata:
      serverAddress: http://prometheus:9090
      metricName: gateway_active_connections_per_replica
      query: |
        sum(rate(gateway_connections_total[1m])) by (pod) / count(pod)
      threshold: "8000"  # 单实例 8k 连接触发扩容
      activationThreshold: "100"
  - type: kafka  # 或 Pulsar
    metadata:
      topic: meeting-permission-cmd
      bootstrapServers: pulsar://...
      consumerGroup: push-gateway
      lagThreshold: "1000"  # 积压 1k 条扩容消费端

# .buf.yaml (Protobuf 变更检测 CI 集成)
version: v1
breaking:
  use:
    - FILE
  ignore:
    - EXTENSION_VALUE # 允许扩展字段新增
  ignore_only:
    - FIELD_SAME_TYPE # 同类型字段号复用视为破坏性变更
lint:
  use:
    - DEFAULT
    - COMMENTS
    - PACKAGE_VERSION_SUFFIX

[美西用户] --> [美西 Edge POP] --> (专线/加速通道) <---> [核心区 集群] <---> (专线) <---> [新加坡 Edge POP] <-- [新加坡用户]
                    |                                                    |
              本地鉴权/路由                                        状态机单一权威源
              权限指令本地回环                                        异步复制到各 Region Redis

# Envoy/NGINX 路由规则：按 Header 或 URL 版本分发
location /ws/v1/ {
    proxy_pass http://gateway-v1-upstream; # 维护模式，仅存量连接
}
location /ws/v2/ {
    proxy_pass http://gateway-v2-upstream; # 主流版本
}
# 透传未知字段：网关层不解码 Protobuf，纯 TCP/QUIC 转发，天然兼容新增字段
# 仅鉴权/路由层解码 Header (meeting_id, uid, version)

# 自愈规则 (基于 Prometheus Alert + Webhook -> Operator)
rules:
- alert: PushGatewayHighConnectionLoad
  expr: gateway_active_connections_per_replica > 9000
  for: 2m
  action:
    type: scale_out
    target: push-gateway
    params:
      delta: +20%
    cooldown: 5m

- alert: PermissionSyncLagHigh
  expr: histogram_quantile(0.99, rate(permission_push_latency_seconds_bucket[5m])) > 1
  for: 3m
  action:
    type: run_playbook
    playbook: permission_lag_mitigation
    steps:
      - check_pulsar_backlog
      - if_backlog_gt_10k: scale_consumer_group
      - check_redis_cpu
      - if_cpu_gt_80: enable_read_replica
      - notify_oncall: "权限推送延迟飙升，已自动扩容消费组/启用只读副本"

- alert: ClientVersionSkew
  expr: count by (sdk_version) (client_online_total) / sum(client_online_total) > 0.05
  for: 1h
  action:
    type: notify
    channel: slack-product
    message: "旧版本 SDK 占比超 5% ({{ $labels.sdk_version }})，建议推送强制升级弹窗"

// 客户端发起连接时注入 Baggage
baggage := "gray_version=v2.3.0,gray_feature=new_perm_model"
ctx = baggage.ContextWithBaggage(ctx, baggage)

// 网关/状态服务/消费者均读取 Baggage，路由至对应灰度实例组
// 日志/链路/指标自动打上 gray_version 标签，Grafana 一键对比灰度/基线指标

-- 原子检查版本号并写入离线消息 (KEYS[1]=offline_list, KEYS[2]=version_key)
-- ARGV[1]=msg_pb_bytes, ARGV[2]=max_len, ARGV[3]=ttl_sec, ARGV[4]=new_version
local cur_ver = tonumber(redis.call('GET', KEYS[2]) or '0')
if tonumber(ARGV[4]) > cur_ver then
    redis.call('LPUSH', KEYS[1], ARGV[1])
    redis.call('LTRIM', KEYS[1], 0, tonumber(ARGV[2])-1)
    redis.call('EXPIRE', KEYS[1], ARGV[3])
    redis.call('SET', KEYS[2], ARGV[4])
    return 1
end
return 0 -- 版本号不增，丢弃

# Linux 内核参数
net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 15
# 关键：开启 TCP_NOTSENT_LOWAT (Kernel 4.1+) 限制未发送队列，降低延迟抖动
net.ipv4.tcp_notsent_lowat = 131072  # 128KB

# Go 网关 Listener 配置
ln, _ := net.ListenConfig{
    Control: func(network, address string, c syscall.RawConn) error {
        return c.Control(func(fd uintptr) {
            syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_NOTSENT_LOWAT, 131072)
            syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_SNDBUF, 2*1024*1024)
        })
    },
}.Listen("tcp", ":8080")

// 统一接口屏蔽 WS/WT 差异
interface IPermissionTransport {
  connect(config: ConnectConfig): Promise<void>;
  disconnect(): void;
  onPermissionChange(handler: (cmd: PermissionChangeCmd) => void): UnsubscribeFn;
  requestSnapshot(sinceVersion: bigint): Promise<PermissionSnapshot>;
  getConnectionQuality(): ConnectionQuality; // RTT、丢包率、协议类型
}

// 业务层仅依赖此接口，通过工厂模式注入 WSClient 或 WTClient
const transport = TransportFactory.create({
  preferWebTransport: true,
  fallbackToWebSocket: true,
  serverEndpoints: ['wss://gateway.example.com', 'https://gateway.example.com'], // WT 用 HTTPS 入口
});