// 核心指标标准化接口示例
interface StandardStats {
  timestamp: number;           // 采集时间戳 (ms)
  peerConnectionId: string;    // PC 实例唯一标识
  // 网络层
  rtt: number;                 // 往返时延
  availableOutgoingBitrate: number; // 可用发送带宽
  // 发送端
  bytesSent: number;
  packetsSent: number;
  packetsLost: number;         // 发送端感知丢包
  // 接收端
  bytesReceived: number;
  packetsReceived: number;
  packetsLost: number;         // 接收端丢包
  jitter: number;              // 抖动
  // 编解码
  codecId: string;
  mimeType: string;            // VP8/H264/OPUS 等
  firCount: number;            // 关键帧请求次数
  pliCount: number;
  // 业务扩展
  userId: string;
  roomId: string;
  streamType: 'main' | 'screen' | 'audio';
}

async function collectStandardStats(pc, context) {
  const report = await pc.getStats();
  const map = new Map(); // id -> RTCStats object
  
  // 1. 扁平化存储，便于 O(1) 查找关联对象
  report.forEach((stat) => map.set(stat.id, stat));

  // 2. 定位核心传输对象
  // 优先查找 candidate-pair (state: succeeded) -> 获取 rtt
  // 关联 local-candidate / remote-candidate -> 获取 IP/协议/网络类型
  
  // 3. 遍历 inbound/outbound rtp 流
  const result = [];
  report.forEach((stat) => {
    if (stat.type === 'outbound-rtp' && stat.kind === 'video') {
      const remoteInbound = map.get(stat.remoteId); // 关联远端接收报告
      const codec = map.get(stat.codecId);
      
      result.push(buildStandardOutbound(stat, remoteInbound, codec, context));
    }
    // ... 处理 inbound-rtp, audio 等同理
  });
  return result;
}

// 简易速率计算器
class RateCalculator {
  constructor() { this.last = new Map(); }
  
  calculate(key, currentValue, currentTime) {
    const last = this.last.get(key);
    this.last.set(key, { value: currentValue, time: currentTime });
    if (!last) return 0; // 首次采集无速率
    const dt = (currentTime - last.time) / 1000; // 秒
    return dt > 0 ? (currentValue - last.value) / dt : 0; // bytes/s
  }
}

class MetricRegistry {
  constructor() {
    this.calculators = new Map(); // name -> (standardStats, context) => value
  }
  
  register(name, calculator) {
    this.calculators.set(name, calculator);
  }
  
  computeAll(standardStats, context) {
    const custom = {};
    this.calculators.forEach((fn, name) => {
      try { custom[name] = fn(standardStats, context); } 
      catch (e) { console.warn(`Custom metric ${name} error:`, e); }
    });
    return custom;
  }
}

// 使用示例：注册一个“视频卡顿判定”指标
registry.register('videoFreezeDetected', (stats, ctx) => {
  // 逻辑：帧间隔 > 500ms 且持续 > 2s 判定为卡顿
  // 需结合历史状态 ctx.history 判断
  return detectFreeze(stats, ctx.history);
});

CREATE TABLE raw_stats_source (
  `timestamp` BIGINT,
  `peer_connection_id` STRING,
  `user_id` STRING,
  `room_id` STRING,
  `stream_type` STRING,
  `rtt` INT,
  `jitter` DOUBLE,
  `packets_lost` BIGINT,
  `packets_received` BIGINT,
  `bytes_received` BIGINT,
  `available_outgoing_bitrate` BIGINT,
  -- 自定义指标以 Map 形式存储，或展平为列
  `custom_metrics` MAP<STRING, STRING>,
  `proc_time` AS PROCTIME(),
  WATERMARK FOR `proc_time` AS `proc_time` - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'webrtc_raw_stats',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json',
  'scan.startup.mode' = 'latest-offset'
);

CREATE VIEW user_stream_quality_1min AS
SELECT
  `user_id`,
  `room_id`,
  `stream_type`,
  TUMBLE_START(`proc_time`, INTERVAL '1' MINUTE) AS window_start,
  -- 网络质量
  AVG(`rtt`) AS avg_rtt,
  MAX(`rtt`) AS max_rtt,
  AVG(`jitter`) AS avg_jitter,
  -- 丢包率计算 (需注意分母为 0)
  CASE WHEN SUM(`packets_received` + `packets_lost`) > 0
       THEN SUM(`packets_lost`) * 1.0 / SUM(`packets_received` + `packets_lost`)
       ELSE 0 END AS packet_loss_rate,
  -- 带宽利用率
  AVG(`bytes_received` * 8 / 1000 / 60) AS avg_recv_kbps, -- 近似分钟平均带宽
  MAX(`available_outgoing_bitrate`) / 1000 AS max_available_kbps,
  -- 自定义指标聚合：卡顿次数求和
  SUM(CAST(`custom_metrics`['videoFreezeDetected'] AS INT)) AS freeze_count,
  COUNT(DISTINCT `peer_connection_id`) AS active_peers
FROM raw_stats_source
GROUP BY
  `user_id`, `room_id`, `stream_type`,
  TUMBLE(`proc_time`, INTERVAL '1' MINUTE);

-- 示例：连续 3 个 1 分钟窗口丢包率 > 10% 触发弱网告警
CREATE TABLE alert_sink (
  alert_id STRING,
  user_id STRING,
  room_id STRING,
  alert_type STRING, -- 'HIGH_PACKET_LOSS', 'HIGH_RTT', 'FREEZE_FREQUENT'
  severity STRING,   -- 'WARN', 'CRITICAL'
  detail STRING,
  trigger_time TIMESTAMP(3)
) WITH (...);

INSERT INTO alert_sink
SELECT
  UUID() AS alert_id,
  `user_id`,
  `room_id`,
  'HIGH_PACKET_LOSS' AS alert_type,
  'WARN' AS severity,
  CONCAT('连续 3 分钟丢包率超阈值, 当前: ', CAST(`packet_loss_rate` * 100 AS STRING), '%') AS detail,
  `window_start` AS trigger_time
FROM (
  SELECT *, 
    COUNT(*) OVER (PARTITION BY `user_id`, `room_id` ORDER BY `window_start` 
                   ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS consecutive_cnt
  FROM user_stream_quality_1min
  WHERE `packet_loss_rate` > 0.1
) t
WHERE consecutive_cnt >= 3;

// main-thread: sdk-core.js
class StatsCollector {
  constructor() {
    this.worker = new Worker('stats-worker.js', { name: 'RTC-Stats-Collector' });
    this.pcMap = new Map(); // pcId -> RTCPeerConnection
    this._bindWorkerEvents();
  }

  registerPeerConnection(pcId, pc) {
    this.pcMap.set(pcId, pc);
    // 仅传递 PC 的标识符，Worker 侧通过 MessageChannel 或 Comlink 代理调用 getStats
    this.worker.postMessage({ type: 'REGISTER_PC', pcId, pcProxy: this._createProxy(pc) }, [pcProxy.port]);
  }

  _createProxy(pc) {
    // 使用 Comlink 或自定义 MessageChannel 封装 RPC
    // 关键：暴露 getStats() 方法供 Worker 远程调用
  }
}

// stats-worker.js (独立线程)
import { RateCalculator, MetricRegistry } from './utils';

const pcProxies = new Map();
const calculator = new RateCalculator();
const registry = new MetricRegistry();

self.onmessage = async (e) => {
  switch (e.data.type) {
    case 'REGISTER_PC':
      pcProxies.set(e.data.pcId, e.data.pcProxy);
      break;
    case 'START_COLLECT':
      startCollectLoop(e.data.interval || 3000);
      break;
  }
};

async function startCollectLoop(interval) {
  while (true) {
    const allStats = [];
    for (const [pcId, proxy] of pcProxies) {
      try {
        const report = await proxy.getStats(); // 跨线程调用
        const standard = transformToStandard(report, pcId, calculator);
        const custom = registry.computeAll(standard);
        allStats.push({ ...standard, custom, pcId });
      } catch (err) { /* 容错：PC 已关闭等 */ }
    }
    if (allStats.length) flushBatch(allStats); // 批量压缩上报
    await sleep(interval);
  }
}

// 弱网对抗策略状态机 (简化版)
class CongestionController {
  constructor() {
    this.state = 'NORMAL'; // NORMAL -> DEGRADE -> RECOVER -> CRITICAL
    this.history = new CircularBuffer(10); // 最近 10 次采集窗口
  }

  feed(features) {
    this.history.push(features);
    const trend = this._analyzeTrend();
    
    // 迟滞比较 防止震荡
    if (this.state === 'NORMAL' && trend.congestion_level >= 2 && trend.loss_rate > 0.15) {
      this.state = 'DEGRADE';
      return this._genActions('DEGRADE', features);
    }
    if (this.state === 'DEGRADE' && trend.congestion_level === 0 && trend.loss_rate < 0.02) {
      this.state = 'RECOVER';
      return this._genActions('RECOVER', features);
    }
    // ... 其他状态迁移
    return null; // 无动作
  }

  _genActions(state, features) {
    const actions = [];
    if (state === 'DEGRADE') {
      // 1. 降码率：目标码率 = 可用带宽 * 0.85 (留余量)
      actions.push({ type: 'SET_BITRATE', target: Math.floor(features.bandwidth_estimate_kbps * 0.85) });
      // 2. 降帧率：15fps -> 10fps
      actions.push({ type: 'SET_FRAMERATE', target: 10 });
      // 3. 强制关键帧：加速新参数生效
      actions.push({ type: 'REQUEST_KEYFRAME' });
      // 4. 开启 FEC/RED (若支持)
      if (features.codec_support_fec) actions.push({ type: 'ENABLE_FEC' });
    }
    return actions;
  }
}

-- 明细表 (MergeTree 分区按天，主键按 room_id, user_id, timestamp)
CREATE TABLE rtc_stats_detail (
    event_time DateTime64(3),
    room_id String,
    user_id String,
    pc_id String,
    stream_type LowCardinality(String), -- main/screen/audio
    -- 核心指标列 (Delta 编码压缩极佳)
    rtt UInt16,
    jitter Float32,
    packet_loss_rate Float32,
    bitrate_kbps UInt32,
    fps UInt8,
    freeze_count UInt16,
    -- 维度列 (LowCardinality 字典编码)
    sdk_version LowCardinality(String),
    browser LowCardinality(String),
    os LowCardinality(String),
    network_type LowCardinality(String), -- wifi/4g/5g/ethernet
    codec LowCardinality(String),
    -- 扩展字段 (Map 存储稀疏自定义指标)
    custom_metrics Map(String, String)
) ENGINE = MergeTree()
PARTITION BY toDate(event_time)
ORDER BY (room_id, user_id, event_time)
TTL event_time + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;

-- 物化视图：分钟级预聚合 (自动维护，查询自动重写)
CREATE MATERIALIZED VIEW rtc_stats_1min_mv
ENGINE = SummingMergeTree()
PARTITION BY toDate(window_start)
ORDER BY (room_id, user_id, stream_type, window_start)
AS SELECT
    toStartOfMinute(event_time) AS window_start,
    room_id, user_id, stream_type,
    count() AS sample_cnt,
    avg(rtt) AS avg_rtt,
    quantile(0.95)(rtt) AS p95_rtt,
    max(jitter) AS max_jitter,
    sum(freeze_count) AS total_freeze,
    -- 状态计数
    countIf(packet_loss_rate > 0.1) AS high_loss_cnt
FROM rtc_stats_detail
GROUP BY window_start, room_id, user_id, stream_type;

// stats-schema.json (CI 门禁)
{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "required": ["timestamp", "pcId", "metrics"],
  "properties": {
    "metrics": {
      "type": "object",
      "properties": {
        "rtt": { "type": "integer", "minimum": 0, "maximum": 5000 },
        "packetLossRate": { "type": "number", "minimum": 0, "maximum": 1 }
      },
      "additionalProperties": false -- 严格模式，禁止未定义字段
    }
  }
}