// statsCollector.ts
interface StandardStats {
  timestamp: number;
  // 核心指标字段...
  inboundRtp: InboundRtpMetrics[];
  outboundRtp: OutboundRtpMetrics[];
  remoteInboundRtp: RemoteInboundRtpMetrics[];
  candidatePair: CandidatePairMetrics[];
  // 扩展字段
  customMetrics: Record<string, any>;
}

class StatsCollector {
  private pc: RTCPeerConnection;
  private timer: number | null = null;
  private buffer: StandardStats[] = [];
  private readonly MAX_BUFFER = 50;

  constructor(pc: RTCPeerConnection) {
    this.pc = pc;
  }

  start(interval = 2000) {
    this.timer = window.setInterval(() => this.poll(), interval);
  }

  stop() {
    if (this.timer) clearInterval(this.timer);
  }

  private async poll() {
    try {
      const report = await this.pc.getStats();
      const standard = this.normalize(report);
      this.pushBuffer(standard);
    } catch (e) {
      console.error('[StatsCollector] poll failed', e);
    }
  }

  // 核心：归一化逻辑
  private normalize(report: RTCStatsReport): StandardStats {
    const result: StandardStats = {
      timestamp: Date.now(),
      inboundRtp: [],
      outboundRtp: [],
      remoteInboundRtp: [],
      candidatePair: [],
      customMetrics: {}
    };

    report.forEach((stat) => {
      switch (stat.type) {
        case 'inbound-rtp':
          if (stat.kind === 'video' || stat.kind === 'audio') {
            result.inboundRtp.push(this.mapInbound(stat));
          }
          break;
        case 'outbound-rtp':
          result.outboundRtp.push(this.mapOutbound(stat));
          break;
        case 'remote-inbound-rtp':
          result.remoteInboundRtp.push(this.mapRemoteInbound(stat));
          break;
        case 'candidate-pair':
          if (stat.nominated) { // 仅关注当前活跃路径
            result.candidatePair.push(this.mapCandidatePair(stat));
          }
          break;
      }
    });

    return result;
  }

  // 字段映射示例：规避 Firefox/Chrome/Safari 字段名不一致
  private mapInbound(s: RTCInboundRtpStreamStats): InboundRtpMetrics {
    return {
      ssrc: s.ssrc,
      kind: s.kind,
      packetsReceived: s.packetsReceived ?? 0,
      packetsLost: s.packetsLost ?? 0,
      jitter: s.jitter ?? 0, // 秒
      framesDecoded: s.framesDecoded ?? 0,
      framesDropped: s.framesDropped ?? 0,
      // 关键：帧率计算需两次采样差值，此处仅存原始值
      // 解码耗时
      totalDecodeTime: s.totalDecodeTime ?? 0,
      // 分辨率 (Chrome 需 track.getSettings())
      frameWidth: s.frameWidth ?? 0,
      frameHeight: s.frameHeight ?? 0,
    };
  }
  // ... mapOutbound, mapCandidatePair 同理
}

// 在 normalize 返回前注入
result.customMetrics = {
  // 业务标签
  roomId: this.context.roomId,
  userId: this.context.userId,
  sdkVersion: '2.4.1',
  platform: navigator.platform,
  // 派生指标 (客户端可计算的简单派生)
  // 例如：瞬时丢包率、瞬时帧率 (需保存上一次采样值)
  instantPacketLossRate: this.calcLossRate(current, last),
  instantFps: this.calcFps(current, last),
  // 设备性能画像
  cpuCores: navigator.hardwareConcurrency,
  deviceMemory: (navigator as any).deviceMemory, // 非标准但常用
};

private flushBuffer() {
  if (this.buffer.length === 0) return;
  
  const payload = {
    // 公共字段提取，减少体积
    common: { roomId: this.context.roomId, userId: this.context.userId },
    // 数据数组
    samples: this.buffer.splice(0, this.buffer.length).map(s => ({
      ts: s.timestamp,
      // 仅上报变化量或关键指标，非全量上报
      in: s.inboundRtp.map(i => ({ ssrc: i.ssrc, pl: i.packetsLost, j: i.jitter * 1000, fps: i.instantFps })),
      out: s.outboundRtp.map(o => ({ ssrc: o.ssrc, br: o.bytesSent, rtt: s.candidatePair[0]?.currentRoundTripTime * 1000 })),
      net: s.candidatePair[0] ? { type: s.candidatePair[0].localCandidateType, rtt: s.candidatePair[0].currentRoundTripTime * 1000 } : null,
      cust: s.customMetrics
    }))
  };

  // 使用 sendBeacon 保证页面卸载也能上报，Content-Type: application/json
  const blob = new Blob([JSON.stringify(payload)], { type: 'application/json' });
  navigator.sendBeacon('/api/v1/rtc/stats/batch', blob);
  
  // 兜底：若 sendBeacon 失败或数据量大，改用 fetch + keepalive
}

-- 1. 原始明细表 (Kafka Engine 消费)
CREATE TABLE rtc_stats_raw
(
    `timestamp` DateTime64(3),
    `room_id` String,
    `user_id` String,
    `sdk_version` String,
    `inbound` Array(Tuple(ssrc String, packets_lost UInt32, jitter_ms Float32, fps Float32, ...)),
    `outbound` Array(Tuple(ssrc String, bytes_sent UInt64, rtt_ms Float32, ...)),
    `network` Tuple(type String, rtt_ms Float32),
    `custom` Map(String, String)
)
ENGINE = Kafka('kafka:9092', 'rtc_stats_topic', 'consumer_group', 'JSONEachRow')
SETTINGS kafka_skip_broken_messages = 100;

-- 2. 明细存储表 (MergeTree, 分区按天，索引 room_id, user_id)
CREATE TABLE rtc_stats_detail
(
    `timestamp` DateTime64(3),
    `room_id` String,
    `user_id` String,
    `kind` Enum8('audio'=1, 'video'=2),
    `direction` Enum8('inbound'=1, 'outbound'=2),
    `ssrc` String,
    `packets_lost` UInt32,
    `jitter_ms` Float32,
    `fps` Float32,
    `rtt_ms` Float32,
    `bytes_sent` UInt64,
    `codec` String,
    `network_type` String,
    -- 物化视图聚合需要的原始字段
    `freeze_count` UInt16 DEFAULT 0,
    `freeze_duration_ms` UInt32 DEFAULT 0
)
ENGINE = MergeTree()
PARTITION BY toDate(timestamp)
ORDER BY (room_id, user_id, timestamp)
TTL toDateTime(timestamp) + INTERVAL 7 DAY DELETE; -- 明细保留 7 天

-- 3. 物化视图：分钟级聚合 (核心大盘数据源)
CREATE MATERIALIZED VIEW rtc_stats_1min_mv
ENGINE = SummingMergeTree()
PARTITION BY toDate(minute)
ORDER BY (room_id, minute, sdk_version, network_type)
AS SELECT
    toStartOfMinute(timestamp) AS minute,
    room_id,
    sdk_version,
    network_type,
    count(DISTINCT user_id) AS user_count,
    -- 网络指标
    avg(rtt_ms) AS avg_rtt_ms,
    quantile(0.95)(rtt_ms) AS p95_rtt_ms,
    sum(packets_lost) / nullif(sum(packets_lost) + sum(packets_received), 0) AS avg_packet_loss_rate,
    -- 视频质量
    avg(fps) AS avg_fps,
    quantile(0.5)(fps) AS median_fps,
    sum(freeze_count) / nullif(sum(user_count), 0) AS avg_freeze_count_per_user,
    sum(freeze_duration_ms) / nullif(sum(play_duration_ms), 0) AS freeze_rate
FROM rtc_stats_detail
WHERE direction = 1 AND kind = 2 -- 仅聚合视频下行
GROUP BY minute, room_id, sdk_version, network_type;

-- 示例：房间级卡顿率突增告警 (分钟级滑动窗口)
SELECT 
    room_id,
    window_start,
    freeze_rate
FROM (
    SELECT 
        room_id,
        HOP_START(minute, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) AS window_start,
        sum(freeze_duration_ms) / nullif(sum(play_duration_ms), 0) AS freeze_rate
    FROM rtc_stats_1min_mv
    GROUP BY room_id, HOP(minute, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE)
)
WHERE freeze_rate > 0.15 -- 卡顿率超过 15% 触发
  AND window_start > now() - INTERVAL '2' MINUTE;

-- ClickHouse 实时计算：带宽预估准确率
SELECT 
    toStartOfMinute(timestamp) AS minute,
    avg(abs(availableOutgoingBitrate - remoteEstimatedMaxBitrate) / nullif(availableOutgoingBitrate, 0)) AS bw_estimation_error_pct,
    quantile(0.95)(currentRoundTripTime) AS p95_rtt
FROM rtc_stats_detail
WHERE direction = 2 -- outbound
GROUP BY minute
HAVING bw_estimation_error_pct > 0.3 -- 偏差超 30% 告警

graph LR
    A[App 发起加入房间] -->|生成 TraceID| B(信令服务器)
    B -->|下发 JoinResponse 携带 TraceID| C[Client SDK]
    C -->|ICE Candidate 携带 TraceID| D[TURN/STUN Server]
    C -->|DTLS/SRTP 握手 携带 TraceID| E[SFU/MCU Media Server]
    E -->|转发媒体流| F[Remote Client]
    C -->|getStats 上报 携带 TraceID| G[Stats Backend]
    E -->|Media Server Stats 上报 携带 TraceID| G
    B -->|信令日志 携带 TraceID| H[Logging System]

{
  "trace_id": "a1b2c3d4e5f6...",          // 全局唯一，W3C TraceContext 标准
  "span_id": "span_123",                  // 当前节点操作 ID
  "parent_span_id": "span_122",           // 调用方 SpanID
  "service_name": "rtc-client-sdk",       // 服务标识
  "resource.attributes": {
    "deployment.environment": "prod",
    "device.model": "iPhone14,2",
    "network.carrier": "CMCC",
    "sdk.version": "3.2.1"
  },
  "events": [
    {"name": "ice_gathering_start", "timestamp": 1699900000.123},
    {"name": "ice_connection_state_connected", "timestamp": 1699900002.456},
    {"name": "first_frame_rendered", "timestamp": 1699900003.789}
  ]
}

// types/rtc-stats.d.ts (建议用工具自动生成自最新 WebIDL)
interface RTCInboundRtpStreamStats extends RTCReceivedRtpStreamStats {
  readonly kind: "audio" | "video";
  readonly packetsLost: number;
  readonly jitter: number; // seconds
  readonly framesDecoded: number;
  readonly framesDropped: number;
  readonly totalDecodeTime: number; // seconds
  readonly qualityLimitationReason?: "none" | "cpu" | "bandwidth" | "other";
  // ... 标准字段
  // 扩展：厂商私有字段需声明
  readonly googFrameRateReceived?: number; // Chrome 私有
}

// 采集器泛型约束
class TypedStatsCollector<T extends StandardStats> {
  private parser: (report: RTCStatsReport) => T;
  constructor(parser: (report: RTCStatsReport) => T) { this.parser = parser; }
  // ...
}

// 核心流程：Pipeline Pattern
interface IPlugin {
  name: string;
  // 生命周期钩子
  onInit?(ctx: Context): void;
  onCollect?(stats: StandardStats, ctx: Context): StandardStats | void; // 可修改/丰富
  onAggregate?(bucket: StatsBucket, ctx: Context): void; // 本地聚合
  onReport?(payload: ReportPayload, ctx: Context): boolean; // 返回 false 阻断上报
  onDestroy?(): void;
}

// 内置插件示例
const plugins: IPlugin[] = [
  { name: 'context-enricher', onCollect: (s, ctx) => { s.custom = { ...s.custom, ...ctx.meta }; }},
  { name: 'derivative-calculator', onCollect: (s, ctx) => { s.custom.instantMos = calcMos(s); }},
  { name: 'privacy-filter', onReport: (p, ctx) => { delete p.custom.deviceId; return true; }}, // 合规兜底
  { name: 'batch-reporter', onReport: (p) => sendBeaconBatch(p) },
];

// 运行时动态加载（支持灰度、远程配置开关）
const pipeline = new PluginPipeline(plugins);
pipeline.execute(rawReport);

// compliance/guard.ts
const FORBIDDEN_KEYS = ['phone', 'email', 'realName', 'idCard', 'gpsLat', 'gpsLng', 'imei', 'imsi', 'macAddr'];
const ABSOLUTE_WORDS = ['零延迟', '绝不卡顿', '完美画质', '行业第一', '顶级', '极致', '永不掉线'];

export function sanitizePayload(payload: any): any {
  // 1. 敏感字段清洗 (深度遍历)
  const clean = JSON.parse(JSON.stringify(payload), (key, value) => {
    if (FORBIDDEN_KEYS.some(k => key.toLowerCase().includes(k))) return '[REDACTED]';
    return value;
  });

  // 2. 文案合规扫描 (针对上报的自定义文案字段)
  if (clean.custom?.errorMsg) {
    ABSOLUTE_WORDS.forEach(w => {
      if (clean.custom.errorMsg.includes(w)) {
        console.warn(`[Compliance] Absolute wording detected: ${w}`);
        clean.custom.errorMsg = clean.custom.errorMsg.replace(w, '优化中');
      }
    });
  }
  return clean;
}

// 在上报管道强制注入
pipeline.use({ name: 'compliance-guard', onReport: (p) => sanitizePayload(p) });