interface UnifiedStats {
  // 通用标识
  timestamp: number;        // 客户端上报时间戳
  sessionId: string;        // 会话 ID
  userId: string;           // 脱敏用户 ID
  peerId: string;           // 对端 ID
  trackId: string;          // MediaStreamTrack ID
  
  // 网络层
  rtt: number;              // 往返时延
  availableOutgoingBitrate: number; // 可用发送带宽
  availableIncomingBitrate: number; // 可用接收带宽
  
  // 发送端指标
  outbound: {
    bytesSent: number;
    packetsSent: number;
    packetsLost: number;        // 累计丢包
    fractionLost: number;       // 丢包率
    roundTripTime: number;      // 发送端感知 RTT
    totalEncodeTime: number;    // 编码总耗时
    framesEncoded: number;
    // ... 分辨率、帧率等
  };
  
  // 接收端指标
  inbound: {
    bytesReceived: number;
    packetsReceived: number;
    packetsLost: number;
    jitter: number;             // 抖动
    jitterBufferDelay: number;  // 抖动缓冲延迟
    totalDecodeTime: number;    // 解码总耗时
    framesDecoded: number;
    framesDropped: number;      // 丢帧数
    // ...
  };
  
  // ICE 状态
  ice: {
    state: 'new' | 'checking' | 'connected' | 'completed' | 'failed' | 'disconnected' | 'closed';
    nominated: boolean;         // 是否被提名
    protocol: 'udp' | 'tcp';    // 传输协议
    candidateType: 'host' | 'srflx' | 'prflx' | 'relay'; // 候选类型
  };
}

// 伪代码：客户端差分计算示例
function calculateDelta(current, previous, intervalMs) {
  if (!previous) return null;
  const dt = (current.timestamp - previous.timestamp) / 1000; // 秒
  return {
    bitrate: (current.bytesSent - previous.bytesSent) * 8 / dt / 1000, // kbps
    packetLossRate: (current.packetsLost - previous.packetsLost) / (current.packetsSent - previous.packetsSent),
    frameRate: (current.framesEncoded - previous.framesEncoded) / dt,
    // ...
  };
}

groups:
- name: webrtc_quality_alerts
  rules:
  # P0: 核心成功率骤降
  - alert: WebRTCCallSuccessRateCritical
    expr: |
      (sum(rate(webrtc_session_success_total[5m])) / sum(rate(webrtc_session_total[5m]))) < 0.95
    for: 5m
    labels:
      severity: p0
      team: rtc-infra
    annotations:
      summary: "WebRTC 通话成功率低于 95%"
      description: "当前成功率: {{ $value | humanizePercentage }}, 请立即检查信令/ICE/TURN 服务状态。"
      runbook_url: "https://wiki.internal/runbook/webrtc-success-rate-drop"

  # P1: 单版本异常 (防止版本回归)
  - alert: WebRTCVersionHighPacketLoss
    expr: |
      sum by (version) (rate(webrtc_packets_lost_total[5m])) 
      / 
      sum by (version) (rate(webrtc_packets_sent_total[5m])) > 0.1
    for: 10m
    labels:
      severity: p1
      team: client-sdk
    annotations:
      summary: "版本 {{ $labels.version }} 丢包率超过 10%"
      description: "疑似新版本 SDK 网络逻辑回归，建议对比上一版本指标或发布灰度回滚。"

// worker.ts - 核心逻辑隔离
const STATS_INTERVAL_MS = 5000;
let prevStatsMap: Map<string, RTCStat> = new Map();

self.onmessage = (e) => {
  if (e.data.type === 'START') {
    const pc = e.data.peerConnection; // 通过 MessageChannel 传递 RTCPeerConnection (需浏览器支持 OffscreenCanvas/Transferable)
    // 若不支持传递 PC，主线程定时调用 getStats 后 postMessage 传原始数据给 Worker 计算
    setInterval(() => collectAndReport(pc), STATS_INTERVAL_MS);
  }
};

async function collectAndReport(pc: RTCPeerConnection) {
  const report = await pc.getStats();
  const currentMap = new Map<string, RTCStat>();
  const deltaMetrics: UnifiedMetrics[] = [];

  report.forEach((stat) => {
    currentMap.set(stat.id, stat);
    const prev = prevStatsMap.get(stat.id);
    if (prev && stat.timestamp > prev.timestamp) {
      const delta = computeDelta(stat, prev); // 纯计算，无 DOM 操作
      if (delta) deltaMetrics.push(delta);
    }
  });

  prevStatsMap = currentMap;

  // 批量压缩上报
  if (deltaMetrics.length) {
    const payload = compress(JSON.stringify(deltaMetrics)); // zstd/wasm 或 pako
    self.postMessage({ type: 'REPORT', payload, byteLength: payload.byteLength });
  }
}

-- 1. 定义源表 (Kafka Source)
CREATE TABLE raw_stats (
    session_id STRING,
    user_id STRING,
    peer_id STRING,
    track_id STRING,
    `timestamp` BIGINT, -- 客户端采集时间
    rtt DOUBLE,
    jitter DOUBLE,
    packet_loss_rate DOUBLE,
    bitrate_kbps DOUBLE,
    frame_rate DOUBLE,
    resolution_width INT,
    resolution_height INT,
    codec STRING,
    ice_state STRING,
    candidate_type STRING,
    -- 水印策略：允许 10s 乱序，基于客户端时间戳
    WATERMARK FOR `ts` AS `ts` - INTERVAL '10' SECOND
) WITH (...);

-- 2. 维表：用户/设备画像 (HBase/Redis/JDBC Lookup)
CREATE TABLE dim_user_profile (
    user_id STRING,
    app_version STRING,
    device_model STRING,
    os_version STRING,
    isp STRING,
    province STRING,
    is_vip BOOLEAN,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc', 'lookup.cache.max-rows' = '100000', 'lookup.cache.ttl' = '1h'
);

-- 3. 宽表输出 (Kafka Sink / Doris Sink)
CREATE TABLE enriched_stats (
    -- 维度标签 (低基数)
    app_version STRING,
    device_model STRING,
    isp STRING,
    province STRING,
    codec STRING,
    track_kind STRING, -- 'video'/'audio'
    -- 指标值
    rtt_ms DOUBLE,
    jitter_ms DOUBLE,
    plr DOUBLE,
    bitrate_kbps DOUBLE,
    fps DOUBLE,
    resolution STRING, -- '1920x1080'
    -- 派生标签
    quality_score DOUBLE, -- 0-100 综合评分
    is_freeze BOOLEAN,    -- 瞬时卡顿判定
    ice_anomaly STRING    -- 'NORMAL'/'RELAY_FALLBACK'/'ICE_RESTART'/'FAILED'
) WITH (...);

-- 4. 核心 ETL 逻辑
INSERT INTO enriched_stats
SELECT
    -- 维表关联 (Async Lookup Join)
    dim.app_version, dim.device_model, dim.isp, dim.province,
    raw.codec,
    CASE WHEN raw.track_id LIKE '%video%' THEN 'video' ELSE 'audio' END as track_kind,
    -- 指标透传
    raw.rtt as rtt_ms,
    raw.jitter as jitter_ms,
    raw.packet_loss_rate as plr,
    raw.bitrate_kbps,
    raw.frame_rate as fps,
    CONCAT(raw.resolution_width, 'x', raw.resolution_height) as resolution,
    -- 派生计算：QoE 评分 (简化版 E-Model)
    GREATEST(0, 100 - raw.rtt * 0.1 - raw.jitter * 2 - raw.packet_loss_rate * 500) as quality_score,
    -- 卡顿判定：帧率 < 5fps 且持续 > 2s (需配合状态编程或 CEP，此处简化为单点阈值)
    CASE WHEN raw.frame_rate < 5 AND raw.track_id LIKE '%video%' THEN TRUE ELSE FALSE END as is_freeze,
    -- ICE 异常标记
    CASE 
        WHEN raw.ice_state = 'failed' THEN 'FAILED'
        WHEN raw.ice_state = 'disconnected' THEN 'ICE_RESTART'
        WHEN raw.candidate_type = 'relay' THEN 'RELAY_FALLBACK'
        ELSE 'NORMAL'
    END as ice_anomaly
FROM raw_stats raw
LEFT JOIN dim_user_profile FOR SYSTEM_TIME AS OF raw.`ts` AS dim
ON raw.user_id = dim.user_id;

-- 会话聚合视图 (1分钟滑动窗口，每 10s 刷新)
CREATE VIEW session_1min_agg AS
SELECT
    session_id,
    -- 维度取任意值 (同一会话维度一致)
    MAX(app_version) as app_version,
    MAX(isp) as isp,
    -- 核心指标聚合
    COUNT(DISTINCT user_id) as participant_cnt,
    AVG(quality_score) as avg_qoe,
    PERCENTILE(rtt_ms, 0.5) as p50_rtt,
    PERCENTILE(rtt_ms, 0.99) as p99_rtt,
    SUM(CASE WHEN is_freeze THEN 1 ELSE 0 END) * 1.0 / COUNT(*) as freeze_rate,
    SUM(CASE WHEN ice_anomaly != 'NORMAL' THEN 1 ELSE 0 END) as ice_issue_cnt,
    -- 通话时长估算 (采样点数 * 采样间隔)
    COUNT(*) * 10 as est_duration_sec
FROM enriched_stats
GROUP BY session_id, HOP(`ts`, INTERVAL '10' SECOND, INTERVAL '1' MINUTE);

-- 实时 Top 10 高丢包版本 (5分钟窗口)
INSERT INTO top_loss_version_sink
SELECT window_start, window_end, app_version, avg_plr
FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY avg_plr DESC) as rn
    FROM (
        SELECT 
            HOP_START(`ts`, INTERVAL '10' SECOND, INTERVAL '5' MINUTE) as window_start,
            HOP_END(`ts`, INTERVAL '10' SECOND, INTERVAL '5' MINUTE) as window_end,
            app_version,
            AVG(plr) as avg_plr
        FROM enriched_stats
        GROUP BY HOP(`ts`, INTERVAL '10' SECOND, INTERVAL '5' MINUTE), app_version
    )
) WHERE rn <= 10;

grafana-dashboards/
├── lib/                    # 通用库
│   ├── common.libsonnet    # 变量、颜色、阈值、通用函数
│   ├── panels.libsonnet    # 标准面板模板 (Stat, Timeseries, Table, Heatmap)
│   └── promql.libsonnet    # 通用 PromQL 片段 (率、分位数、TopK)
├── dashboards/
│   ├── overview/           # 总览大盘
│   │   ├── main.jsonnet
│   │   └── panels/         # 总览专用面板
│   ├── quality/            # 质量深度大盘
│   ├── network/            # 网络诊断大盘
│   └── session/            # 单会话钻取大模板
├── overlays/               # 环境差异
│   ├── prod.jsonnet
│   └── staging.jsonnet
└── Makefile / CI 脚本      # 编译 -> 校验 -> 部署

// lib/panels.libsonnet
local common = import 'common.libsonnet';

{
  // 标准化 Time Series 面板
  makeTimeseries(title, targets, unit='short', overrides=[]) :: {
    type: 'timeseries',
    title: title,
    gridPos: { h: 8, w: 12, x: 0, y: 0 }, // 由布局引擎自动计算
    targets: targets,
    fieldConfig: {
      defaults: {
        unit: unit,
        thresholds: common.defaultThresholds, // 绿<黄<红通用阈值
        custom: { lineWidth: 1, fillOpacity: 10, drawStyle: 'line' }
      },
      overrides: overrides + [
        // 自动为包含 "p99" 的系列加粗虚线
        { matcher: { type: 'byName', options: 'p99' }, properties: [{ id: 'drawStyle', value: 'dashes' }, { id: 'lineWidth', value: 2 }] }
      ]
    }
  },

  // 标准化 Stat 单值面板 (含 Sparkline 趋势图)
  makeStat(title, target, unit, thresholds=common.slaThresholds) :: {
    type: 'stat',
    title: title,
    targets: [target],
    fieldConfig: {
      defaults: {
        unit: unit,
        thresholds: thresholds,
        mappings: [{ type: 'value', options: { '0': '✅ 正常', '1': '⚠️ 预警', '2': '🔥 故障' } }] // 状态码映射
      }
    },
    options: { graphMode: 'area', justifyMode: 'auto', colorMode: 'value' }
  },
}

// lib/common.libsonnet
{
  standardVariables: [
    { name: 'datasource', type: 'datasource', label: '数据源', default: 'VictoriaMetrics-Prod', query: 'prometheus' },
    { name: 'app_id', type: 'query', label: '应用', query: 'label_values(webrtc_session_total, app_id)', multi: true, includeAll: true, refresh: 1 },
    { name: 'version', type: 'query', label: '版本', query: 'label_values(webrtc_session_total{app_id=~"$app_id"}, version)', multi: true, includeAll: true, refresh: 2 },
    { name: 'isp', type: 'query', label: '运营商', query: 'label_values(webrtc_session_total{app_id=~"$app_id"}, isp)', multi: true, includeAll: true },
    { name: 'interval', type: 'interval', label: '聚合粒度', default: '1m', options: ['10s','30s','1m','5m','15m'] },
  ],

  // Drilldown 链接生成函数
  makeDrilldownLink(baseUrl, timeRange='1h') :: 
    '${baseUrl}?var-app_id=${__value.raw.app_id}&var-version=${__value.raw.version}&var-session_id=${__value.raw.session_id}&from=now-${timeRange}&to=now',
}

# .gitlab-ci.yml / GitHub Actions
stages: [validate, build, deploy]

validate_dashboards:
  stage: validate
  script:
    - jsonnet -J lib dashboards/overview/main.jsonnet > /dev/null  # 语法校验
    - grafana-toolkit validate dashboards/**/*.json  # Schema 校验 (Panel ID 冲突、Datasource UID 存在性)

build_dashboards:
  stage: build
  script:
    - mkdir -p dist
    - for f in dashboards/*/main.jsonnet; do jsonnet -J lib $f > dist/$(basename $(dirname $f)).json; done
  artifacts:
    paths: [dist/]

deploy_grafana:
  stage: deploy
  needs: [build_dashboards]
  script:
    - grafana-http-client import --folder="WebRTC Monitoring" --input=dist/*.json --overwrite
  rules:
    - if: $CI_COMMIT_BRANCH == "main"

# SDK 侧采样决策伪代码
def should_sample(session_context):
    # 1. 强制全采集白名单
    if session_context.user_id in VIP_SET or session_context.version in CANARY_VERSIONS:
        return True, 1.0 # 采样率 100%
    
    # 2. 基于 Session ID 一致性哈希采样 (保证同一会话采样决策一致)
    hash_val = mmh3.hash64(session_context.session_id)[0] % 10000
    
    # 3. 动态基础采样率 (由下发配置控制，默认 10%)
    base_rate = get_remote_config('stats_sample_rate', 1000) # 1000/10000 = 10%
    
    # 4. 异常触发增强采样 (本地判断)
    if session_context.current_plr > 0.05 or session_context.current_rtt > 300:
        return True, 1.0 # 异常期全采集
        
    return hash_val < base_rate, base_rate / 10000.0

-- 查询时按采样率加权还原真实总量
SELECT 
    SUM(bitrate_kbps * sample_weight) as total_bitrate_kbps,
    SUM(packet_loss * sample_weight) / SUM(packets_sent * sample_weight) as weighted_plr
FROM enriched_stats
WHERE sample_weight > 0 -- sample_weight = 1.0 / sample_rate