┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│  客户端 SDK  │◄───►│  接入网关集群  │◄───►│  会议状态服务    │
│ (状态机+UI) │     │ (WS/HTTP2)   │     │ (权限计算+持久化) │
└─────────────┘     └──────────────┘     └────────┬────────┘
                                                   │
                                            ┌──────▼──────┐
                                            │ Redis Pub/Sub │
                                            │  (广播分发)   │
                                            └─────────────┘

{
  "event": "permission.update",
  "meetingId": "m_7x9k2p",
  "timestamp": 1724567890123,
  "version": 42,
  "changes": [
    {
      "userId": "u_8832",
      "role": "co_host",
      "permissions": {
        "mute_others": true,
        "record": true,
        "lock_meeting": false
      },
      "reason": "promote_by_host"
    },
    {
      "userId": "u_1024",
      "permissions": {
        "screen_share": false
      },
      "reason": "revoke_by_cohost"
    }
  ]
}

enum PermissionAction {
  MUTE_OTHERS = 'mute_others',
  RECORD = 'record',
  SCREEN_SHARE = 'screen_share',
  LOCK_MEETING = 'lock_meeting',
  KICK_USER = 'kick_user'
}

interface PermissionState {
  role: 'host' | 'co_host' | 'speaker' | 'audience';
  permissions: Record<PermissionAction, boolean>;
  version: number;
}

class PermissionStore {
  private state: PermissionState;
  private subscribers: Set<(state: PermissionState) => void> = new Set();

  applyPatch(patch: PermissionPatch) {
    if (patch.version <= this.state.version) return; // 乱序丢弃
    this.state = {
      ...this.state,
      ...patch.changes.reduce((acc, c) => ({
        ...acc,
        permissions: { ...acc.permissions, ...c.permissions },
        role: c.role ?? acc.role
      }), this.state),
      version: patch.version
    };
    this.notify();
  }

  subscribe(fn) { this.subscribers.add(fn); }
  unsubscribe(fn) { this.subscribers.delete(fn); }
  private notify() { this.subscribers.forEach(fn => fn(this.state)); }
}

function usePermission(action: PermissionAction) {
  const store = usePermissionStore();
  return useSyncExternalStore(
    store.subscribe,
    () => store.state.permissions[action],
    () => false // SSR 回退值
  );
}

// 业务组件
function RecordButton() {
  const canRecord = usePermission(PermissionAction.RECORD);
  return <Button disabled={!canRecord} onClick={startRecord}>录制</Button>;
}

# permissions.yaml
roles:
  host:
    base: [mute_others, record, lock_meeting, kick_user, screen_share]
  co_host:
    base: [mute_others, record, screen_share]
    deny: [lock_meeting, kick_user]
  speaker:
    base: [screen_share]
  audience:
    base: []

@Transactional
public PermissionSnapshot changeRole(ChangeRoleCmd cmd) {
    Meeting meeting = meetingRepo.lockById(cmd.meetingId); // SELECT FOR UPDATE
    long expectedVersion = cmd.clientVersion;
    if (meeting.getVersion() != expectedVersion) {
        throw new ConflictException("版本冲突，请刷新重试");
    }
    // 计算新权限...
    meeting.setVersion(meeting.getVersion() + 1);
    meetingRepo.save(meeting);
    eventPublisher.publish(buildEvent(meeting));
    return meeting.toSnapshot();
}

public final class PermissionCodec extends ByteToMessageCodec<PermissionFrame> {

    private static final int HEADER_LEN = 12; // magic(2) + version(1) + type(1) + length(4) + traceId(4)

    @Override
    protected void encode(ChannelHandlerContext ctx, PermissionFrame msg, ByteBuf out) {
        // 1. 预计算 payload 大小，避免扩容
        byte[] payload = ProtobufSerializer.serialize(msg.getPayload());
        int capacity = HEADER_LEN + payload.length;
        ByteBuf buffer = ctx.alloc().ioBuffer(capacity);
        
        // 2. 直接写入头部
        buffer.writeShort(MAGIC).writeByte(PROTOCOL_VERSION).writeByte(msg.getType())
            .writeInt(payload.length).writeInt(msg.getTraceId());
        
        // 3. 零拷贝写入 payload
        buffer.writeBytes(payload);
        out.writeBytes(buffer);
        buffer.release(); // 仅释放引用计数，底层内存由 payload 共享
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < HEADER_LEN) return;
        in.markReaderIndex();
        // 校验魔数、版本、长度合法性...
        int length = in.readInt();
        if (in.readableBytes() < length) { in.resetReaderIndex(); return; }
        
        // 关键：retainedSlice 零拷贝切片，传给业务线程池
        ByteBuf payloadSlice = in.readRetainedSlice(length);
        PermissionFrame frame = parseFrame(in, payloadSlice);
        out.add(frame);
    }
}

@Component
public class SessionRegistry {

    // meetingId -> Channel 集合 (Netty Channel 是线程安全的)
    private final ConcurrentHashMap<Long, ConcurrentHashSet<Channel>> meetingChannels = new ConcurrentHashMap<>();
    // userId -> Channel 集合 (支持多端)
    private final ConcurrentHashMap<Long, ConcurrentHashSet<Channel>> userChannels = new ConcurrentHashMap<>();

    public void bindMeeting(long meetingId, Channel channel) {
        meetingChannels.computeIfAbsent(meetingId, k -> new ConcurrentHashSet<>()).add(channel);
    }

    public void unbindMeeting(long meetingId, Channel channel) {
        ConcurrentHashSet<Channel> set = meetingChannels.get(meetingId);
        if (set != null) {
            set.remove(channel);
            if (set.isEmpty()) meetingChannels.remove(meetingId, set); // 原子清理空集合
        }
    }

    // 广播时无锁快照迭代
    public void broadcast(long meetingId, PermissionFrame frame) {
        ConcurrentHashSet<Channel> channels = meetingChannels.get(meetingId);
        if (channels == null) return;
        // 并行写入，利用 Netty EventLoop 保序
        channels.forEach(ch -> ch.writeAndFlush(frame).addListener(f -> {
            if (!f.isSuccess()) handleWriteFailure(ch, f.cause());
        }));
    }
}

-- KEYS[1]: meeting:channel:{meetingId} (Stream)
-- KEYS[2]: gateway:subscribed:{gatewayId} (Set)
-- ARGV[1]: gatewayId
-- ARGV[2]: lastConsumedId (客户端上报版本)
-- 返回: {新消息数组, 最新ID}

local streamKey = KEYS[1]
local subKey = KEYS[2]
local gatewayId = ARGV[1]
local lastId = ARGV[2]

-- 1. 原子记录订阅关系
redis.call('SADD', subKey, gatewayId)

-- 2. 读取增量消息 (阻塞 0ms 即非阻塞)
local messages = redis.call('XRANGE', streamKey, '(' .. lastId, '+', 'COUNT', 1000)

-- 3. 返回最新 ID 供客户端下次上报
local latestId = redis.call('XREVRANGE', streamKey, '+', '-', 'COUNT', 1)
if #latestId > 0 then latestId = latestId[1][1] else latestId = '0-0' end

return {messages, latestId}

{
  "event_id": "evt_7x9k2p_42",
  "timestamp": "2024-08-25T10:30:00.123Z",
  "meeting_id": "m_7x9k2p",
  "operator": { "user_id": "u_1001", "role": "host", "ip": "203.0.113.45", "device_fp": "fp_abc..." },
  "target": { "user_id": "u_8832", "prev_role": "speaker", "new_role": "co_host" },
  "permission_delta": { "mute_others": [false, true], "record": [false, true] },
  "reason": "promote_by_host",
  "client_version": "5.2.1",
  "trace_id": "trace_9f2a1c",
  "compliance_tags": ["role_escalation", "data_access_change"]
}

// 基于 ObjectPool (Apache Commons Pool 2) 封装
public final class FramePool {
    private static final GenericObjectPool<PermissionFrame> POOL = new GenericObjectPool<>(
        new BasePooledObjectFactory<>() {
            @Override public PermissionFrame create() { return new PermissionFrame(); }
            @Override public void passivateObject(PooledObject<PermissionFrame> p) { p.getObject().reset(); }
        },
        new GenericObjectPoolConfig<>() {{
            setMaxTotal(20000); setMinIdle(500); setTestOnBorrow(false);
        }}
    );

    public static PermissionFrame acquire() { try { return POOL.borrowObject(); } catch (Exception e) { return new PermissionFrame(); } }
    public static void release(PermissionFrame f) { POOL.returnObject(f); }
}

// 业务处理器
@ChannelHandler.Sharable
public class PermissionHandler extends SimpleChannelInboundHandler<PermissionFrame> {
    @Override protected void channelRead0(ChannelHandlerContext ctx, PermissionFrame frame) {
        try { process(frame); } finally { FramePool.release(frame); } // 必须 finally 释放
    }
}

// 服务端聚合逻辑
public void batchMuteAll(long meetingId, List<Long> targetUsers, boolean mute) {
    PermissionFrame frame = PermissionFrame.builder()
        .type(FrameType.BATCH_PERM_UPDATE)
        .meetingId(meetingId)
        .payload(BatchPermUpdate.newBuilder()
            .addAllTargets(targetUsers.stream().map(uid -> 
                PermDelta.newBuilder().setUserId(uid).putPerms("mute", mute).build()
            ).collect(toList()))
            .build())
        .build();
    // 单次广播替代 N 次单播
    gatewayService.broadcast(meetingId, frame); 
}

# pact/contracts/permission_update.yaml
provider: "meeting-gateway"
consumer: "web-sdk"
interactions:
  - description: "权限增量推送"
    request:
      method: "WS"
      path: "/ws/v1/meeting/m_7x9k2p"
      headers: { "Sec-WebSocket-Protocol": "protobuf.permission.v1" }
    response:
      status: 101
      body:
        event: "permission.update"
        meetingId: "m_7x9k2p"
        version: 42
        changes:
          - userId: "u_8832"
            role: "co_host"
            permissions:
              mute_others: true
              record: true
    metadata:
      contentType: "application/x-protobuf"

# 灰度规则配置中心
canary_rules:
  - feature: "new_permission_engine_v2"
    rollout: 10%  # 按 meetingId % 100 < 10
    metrics_guard:
      - metric: "permission_push_latency_p99"
        threshold: 100ms
      - metric: "permission_conflict_rate"
        threshold: 0.5%
    auto_rollback: true

@SentinelResource(value = "changeRole", 
    blockHandler = "fallbackChangeRole",
    fallback = "defaultChangeRole")
public Response changeRole(ChangeRoleCmd cmd) {
    // 核心逻辑...
}

// 熔断降级：返回本地缓存权限，禁止写入
public Response fallbackChangeRole(ChangeRoleCmd cmd, BlockException ex) {
    log.warn("权限变更熔断，触发降级", ex);
    return Response.throttled("系统繁忙，请稍后重试");
}

// 兜底：仅读取本地内存快照，不走 DB/Redis
public Response defaultChangeRole(ChangeRoleCmd cmd, Throwable ex) {
    log.error("权限变更异常兜底", ex);
    return Response.degraded("当前仅支持查看权限，修改功能暂不可用");
}

// 握手协议
message HandshakeReq {
  string sdk_version = 1;      // "5.2.1"
  repeated int32 supported_proto_versions = 2; // [1, 2, 3]
  string device_id = 3;
}

message HandshakeResp {
  int32 negotiated_version = 1; // 服务端选定版本
  string server_time = 2;       // 时间同步
  repeated FeatureFlag features = 3; // 新功能开关
}