┌─────────────────────────────────────────────────────────────┐
│                      接入层 (Gateway)                        │
│  WebSocket / WebRTC DataChannel / HTTP Long-polling          │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                      协同状态服务层                           │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐           │
│  │ 投票聚合引擎 │  │ 问答排序引擎 │  │ 会话状态机   │           │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘           │
└─────────┼────────────────┼────────────────┼──────────────────┘
          │                │                │
┌─────────▼────────────────▼────────────────▼──────────────────┐
│                    高并发消息总线层                            │
│  ┌─────────────────────────────────────────────────────────┐  │
│  │  主题：meeting.{meetingId}.poll / .qa / .state           │  │
│  │  分区键：questionId / userId（保证同一题目有序）          │  │
│  └─────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                      持久化与索引层                           │
│  Redis Cluster（热数据/计数器） + ClickHouse（全量日志/分析）  │
└─────────────────────────────────────────────────────────────┘

-- KEYS[1] = poll:{meetingId}:{questionId}:counter
-- ARGV[1] = optionId, ARGV[2] = userId, ARGV[3] = ttlSec
local voted = redis.call('SISMEMBER', KEYS[1]..':voters', ARGV[2])
if voted == 1 then return 0 end  -- 幂等去重
redis.call('HINCRBY', KEYS[1], ARGV[1], 1)
redis.call('SADD', KEYS[1]..':voters', ARGV[2])
redis.call('EXPIRE', KEYS[1], ARGV[3])
redis.call('EXPIRE', KEYS[1]..':voters', ARGV[3])
return 1

def hot_score(like_cnt: int, reply_cnt: int, create_ts: int, now: int) -> float:
    """
    改进版 Reddit Hot Ranking，引入时间衰减与交互权重
    """
    age_hours = (now - create_ts) / 3600.0
    base = math.log10(max(like_cnt * 2 + reply_cnt * 3, 1))
    return round(base / (age_hours + 2) ** 1.5, 4)

flowchart TD
    A[开始选型] --> B{是否需要多协议接入<br/>MQTT/AMQP/WebSocket?}
    B -- 是 --> C[Pulsar / NATS]
    B -- 否 --> D{团队运维能力<br/>是否精简?}
    D -- 弱/精简 --> E[NATS JetStream]
    D -- 强 --> F{是否有金融级事务/严格顺序需求?}
    F -- 是 --> G[RocketMQ]
    F -- 否 --> H{现有技术栈是否已深度绑定 Kafka?}
    H -- 是 --> I[Kafka + KRaft]
    H -- 否 --> J[NATS JetStream 推荐]

jetstream {
  max_memory: 64GB
  max_file: 512GB
  store_dir: "/data/jetstream"
}
# 主题策略
subjects {
  "meeting.>.poll.*"   { max_msg_size: 4KB, max_msgs: 1M, retention: limits }
  "meeting.>.qa.*"     { max_msg_size: 8KB, max_msgs: 2M, retention: limits }
  "meeting.>.state.*"  { max_msg_size: 1KB, max_msgs: 500K, retention: workqueue }
}
# 消费者：每会议一个 Durable Consumer，AckPolicy=Explicit, MaxAckPending=5000

sequenceDiagram
    participant Host as 主持人端
    participant GW as 接入网关
    participant JS as NATS JetStream
    participant Agg as 聚合引擎
    participant Redis as Redis Cluster
    participant Att as 参会者端

    Host->>GW: HTTP POST /polls (题目+选项)
    GW->>JS: Publish meeting.{mid}.poll.create (持久化)
    JS-->>Agg: Push 消息
    Agg->>Redis: 初始化计数器 Hash + Set (TTL=会议时长+1h)
    Agg->>JS: Publish meeting.{mid}.poll.{qid}.start
    JS-->>Att: Push 实时题目卡片
    Att->>GW: WebSocket 发送 vote:{qid, optId}
    GW->>JS: Publish meeting.{mid}.poll.{qid}.vote (分区键=qid)
    JS-->>Agg: 批量消费 (BatchSize=200, MaxWait=50ms)
    Agg->>Redis: Lua 原子计数 + 去重
    Agg->>JS: Publish meeting.{mid}.poll.{qid}.result (聚合后数据)
    JS-->>Att: 广播最新票数分布

// SDK 内部实现，业务代码无感
public class ShardedCounter {
    private static final int SHARD_NUM = 16; // 2 的幂，便于位运算
    private final RedisClusterTemplate redis;
    private final String baseKey; // poll:{mid}:{qid}

    public long increment(String optionId, String userId) {
        // 1. 幂等去重：用户级别全局去重键，不分片，体积小
        String dedupKey = baseKey + ":voters";
        Boolean isNew = redis.opsForSet().add(dedupKey, userId);
        if (Boolean.FALSE.equals(isNew)) return 0L; // 重复投票

        // 2. 分片写入：userId hash 打散到 16 个子计数器
        int shard = Math.abs(userId.hashCode()) & (SHARD_NUM - 1);
        String shardKey = baseKey + ":cnt:" + shard;
        redis.opsForHash().increment(shardKey, optionId, 1);
        return 1L;
    }

    // 3. 合并读：Pipeline 并行拉取 16 个分片，本地聚合
    public Map<String, Long> getAllOptions() {
        List<String> keys = IntStream.range(0, SHARD_NUM)
            .mapToObj(i -> baseKey + ":cnt:" + i).toList();
        List<Map<Object, Object>> results = redis.executePipelined(
            (RedisCallback<Object>) conn -> {
                for (String k : keys) conn.hGetAll(k.getBytes());
                return null;
            });
        // 本地合并
        return results.stream()
            .flatMap(m -> m.entrySet().stream())
            .collect(Collectors.groupingBy(
                e -> (String) e.getKey(),
                Collectors.summingLong(e -> ((Number) e.getValue()).longValue())
            ));
    }
}

CREATE TABLE events.poll_qa_detail
(
    event_time      DateTime64(3) CODEC(Delta, ZSTD(3)),
    meeting_id      UInt64 CODEC(ZSTD(3)),
    question_id     UInt64 CODEC(ZSTD(3)),
    user_id         UInt64 CODEC(ZSTD(3)),
    event_type      Enum8('vote'=1, 'like'=2, 'ask'=3, 'pin'=4, 'reply'=5) CODEC(ZSTD(3)),
    option_id       UInt16 CODEC(ZSTD(3)),          -- 投票选项
    content_hash    UInt64 CODEC(ZSTD(3)),          -- 问答内容去重指纹 (SimHash)
    device_type     LowCardinality(String) CODEC(ZSTD(3)),
    network_type    LowCardinality(String) CODEC(ZSTD(3)),
    latency_ms      UInt16 CODEC(ZSTD(3)),          -- 端到端延迟
    is_abnormal     UInt8 CODEC(ZSTD(3))            -- 风控标记
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(event_time)
ORDER BY (meeting_id, question_id, event_time, user_id)
TTL event_time + INTERVAL 7 DAY TO DISK 'hdd',       -- 热数据 SSD
    event_time + INTERVAL 90 DAY TO VOLUME 's3_cold', -- 温数据对象存储
    event_time + INTERVAL 3 YEAR DELETE               -- 合规保留期满删除
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;

// SDK 核心数据结构 (TypeScript 伪代码)
interface VoteState {
  // G-Counter: 每个选项一个单调递增计数器，Key = optionId, Value = {replicaId: count}
  counters: Map<string, Map<string, number>>; 
  // LWW-Element-Set: 记录“我投了哪个选项”，解决“撤销/改投”语义
  // 元素: {optionId, timestamp, replicaId} -> 取 timestamp 最大者为准
  myVote: LWWElementSet<{optionId: string}>; 
}

// 合并函数：纯函数，无副作用，可在 Web Worker 离线执行
function mergeVoteState(local: VoteState, remote: VoteState): VoteState {
  return {
    counters: mergeGCounters(local.counters, remote.counters), // 取每副本最大值
    myVote: mergeLWW(local.myVote, remote.myVote)              // 取时间戳最大
  };
}

graph LR
    Client[Client SDK] -->|QUIC/UDP| LB[L4 LB (支持 UDP)]
    LB --> GW[Gateway (ngx_quic / envoy quic)]
    GW -->|HTTP/3| NATS[NATS JetStream]
    GW -->|gRPC| Agg[Aggregator]

# 多阶段构建：Distroless 基础镜像，启动 < 500ms
FROM golang:1.22-alpine AS builder
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 go build -ldflags="-s -w" -o /aggregator .

FROM gcr.io/distroless/static-debian12:nonroot
COPY --from=builder /aggregator /
USER nonroot:nonroot
ENTRYPOINT ["/aggregator"]

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: poll-aggregator-scaler
  namespace: meeting-prod
spec:
  scaleTargetRef:
    name: poll-aggregator # Knative Service 生成的 Deployment
  pollingInterval: 5 # 5 秒评估一次
  cooldownPeriod: 120 # 2 分钟冷却
  minReplicaCount: 0 # 允许缩容到 0
  maxReplicaCount: 200
  advanced:
    restoreToOriginalReplicaCount: false
    horizontalPodAutoscalerConfig:
      behavior:
        scaleUp:
          stabilizationWindowSeconds: 10
          policies:
            - type: Percent
              value: 100
              periodSeconds: 10
            - type: Pods
              value: 10
              periodSeconds: 10
          selectPolicy: Max
        scaleDown:
          stabilizationWindowSeconds: 60
  triggers:
    - type: nats
      metadata:
        server: "nats://nats-js.meeting.svc:4222"
        subject: "meeting.>.poll.*.vote" # 通配符匹配所有会议投票主题
        queueGroup: "poll-aggregator-group"
        # 核心指标：每个 Pod 处理 500 条积压消息
        lagThreshold: "500" 
        # 可选：按会议维度细粒度扩缩容 (需 NATS 监控暴露 per-consumer lag)
        # account: "MEETING"

用户就近接入 (GSLB / HTTPDNS)
       │
       ├── 华东 1 (Primary) ──────────────────┐
       │  Gateway + Aggregator + NATS + Redis  │  同城双活 (跨 AZ, < 2ms)
       │  ClickHouse (Sync Replica)            │  RPO=0, RTO<30s
       └── 华东 2 (Standby) ──────────────────┘
       │
       ├── 华北 1 (Read Replica / DR) ────────┐
       │  Gateway (只读) + Redis (Async Rep)   │  异地多活 (跨 Region, ~30ms)
       │  ClickHouse (Async Replica)           │  RPO<1s, RTO<5min
       └── 海外 (Singapore) ──────────────────┘

// 统一审计事件结构 (Protobuf v3)
message AuditEvent {
  string event_id = 1;          // UUID v7 (时间有序)
  int64 timestamp_ms = 2;       // 事件发生时间
  string actor_id = 3;          // 操作者 userId / system
  string actor_role = 4;        // HOST / ATTENDEE / ADMIN / SYSTEM
  string action = 5;            // CREATE_POLL / VOTE / LIKE / EXPORT_DATA
  string resource_type = 6;     // POLL / QUESTION / MEETING
  string resource_id = 7;
  map<string, string> context = 8; // {meeting_id, client_ip, device_fingerprint, risk_level}
  // 关键：结果哈希链，防篡改
  string prev_event_hash = 9;   // 上一条事件的 Hash
  string payload_hash = 10;     // 业务载体 Hash (SHA-256)
}

flowchart TD
    UserInput[用户提问/投票选项文本] --> Gateway
    Gateway --> SyncCheck[同步检测 < 50ms]
    SyncCheck -->|高危/确定违规| Block[拦截返回错误码 4003]
    SyncCheck -->|低风险/不确定| AsyncQueue[异步检测队列 NATS]
    AsyncQueue --> AIModel[多模型融合: 文本分类+BERT+规则引擎]
    AIModel -->|违规| Callback[回调业务: 标记隐藏/撤回]
    AIModel -->|疑似| HumanReview[人工复审工单系统]
    HumanReview -->|确认违规| Callback
    HumanReview -->|误判| Whitelist[加入白名单/模型微调]