graph LR
    A[MediaStreamTrack] --> B(Encoder)
    B --> C[RTCRtpSender Internal Pipeline]
    C --> D[Insertable Streams: readable]
    D --> E[TransformStream / JS Logic]
    E --> F[Insertable Streams: writable]
    F --> G[Packetizer -> SRTP -> Network]
    
    G --> H[Network]
    H --> I[SRTP Decrypt -> Depacketizer]
    I --> J[RTCRtpReceiver Internal Pipeline]
    J --> K[Insertable Streams: readable]
    K --> L[TransformStream / JS Logic]
    L --> M[Insertable Streams: writable]
    M --> N[Decoder -> MediaStreamTrack]

// metadata-injector.js
class FrameMetadataInjector {
  constructor() {
    this.frameCounter = 0;
  }

  // 核心转换逻辑
  transform(encodedFrame, controller) {
    // 1. 仅处理视频关键帧或所有帧（视业务而定）
    if (encodedFrame.type !== 'key' && encodedFrame.type !== 'delta') {
      controller.enqueue(encodedFrame);
      return;
    }

    // 2. 构造元数据载荷 (二进制结构: uint32 frameId + uint64 serverTimestamp)
    // 实际项目建议使用 Protocol Buffers 或 MessagePack 序列化
    const metadata = new ArrayBuffer(12);
    const view = new DataView(metadata);
    view.setUint32(0, this.frameCounter++, true); // Little Endian
    view.setBigUint64(4, BigInt(Date.now()), true); // 模拟服务端时间戳

    // 3. 关键步骤：将元数据附加到帧对象
    // 注意：metadata 属性在规范中用于承载不进入 RTP Payload 的辅助数据
    // 若需写入 RTP Header Extension，请见第 5 节
    const newFrame = new RTCEncodedVideoFrame({
      type: encodedFrame.type,
      timestamp: encodedFrame.timestamp,
      data: encodedFrame.data, // 原始编码数据 (H.264/VP8/VP9/AV1 NALUs)
      metadata: metadata       // <--- 注入点
    });

    // 4. 传递给下游
    controller.enqueue(newFrame);
  }
}

async function setupSenderWithMetadata(pc, track) {
  // 1. 创建 Sender
  const sender = pc.addTrack(track);

  // 2. 启用 Insertable Streams (需在 createOffer/createAnswer 前或配置中指定)
  // 标准写法：pc = new RTCPeerConnection({ encodedInsertableStreams: true });
  
  // 3. 获取可读/可写流
  const { readable, writable } = sender.createEncodedStreams();

  // 4. 构建管道: readable -> transformer -> writable
  const transformer = new TransformStream(new FrameMetadataInjector());
  
  // 管道连接 (背压自动处理)
  readable.pipeThrough(transformer).pipeTo(writable).catch(console.error);

  return sender;
}

a=extmap:10 urn:ietf:params:rtp-hdrext:toffset
a=extmap:11 http://example.com/extension/frame-id
a=extmap:12 http://example.com/extension/ai-tags

// rtp-extension-writer.js
const EXT_ID_FRAME_ID = 11; // 对应 SDP 中定义的 ID
const EXT_ID_AI_TAGS  = 12;

class RtpExtensionWriter {
  transform(frame, controller) {
    if (!frame || frame.type === 'empty') { // 处理填充帧
      controller.enqueue(frame);
      return;
    }

    try {
      // 1. 写入帧 ID (Uint16)
      const frameId = this.getNextFrameId(); // 业务逻辑生成
      frame.setHeaderExtension(EXT_ID_FRAME_ID, new Uint16Array([frameId]).buffer);

      // 2. 写入 AI 标签 (变长字符串/二进制，需注意 MTU 限制)
      // RTP 头扩展总长度受限，建议单扩展不超过 100-200 字节
      const aiTags = JSON.stringify({ face: 1, motion: 0.8 });
      const encoder = new TextEncoder();
      frame.setHeaderExtension(EXT_ID_AI_TAGS, encoder.encode(aiTags).buffer);

    } catch (e) {
      console.warn('RTP Header Extension write failed:', e);
      // 部分浏览器/编码器不支持动态设置扩展，需降级处理
    }

    controller.enqueue(frame);
  }
  
  getNextFrameId() { /* ... */ }
}

// receiver-side.js
async function setupReceiverExtensions(receiver) {
  const { readable, writable } = receiver.createEncodedStreams();

  const reader = readable.getReader();
  const writer = writable.getWriter();

  // 使用 ReadableStreamDefaultReader 手动控制读取循环 (更灵活)
  // 或使用 TransformStream (更标准)
  const processor = new TransformStream({
    transform(frame, controller) {
      // 1. 读取帧 ID
      try {
        const frameIdBuf = frame.getHeaderExtension(EXT_ID_FRAME_ID);
        if (frameIdBuf) {
          const frameId = new DataView(frameIdBuf).getUint16(0, true);
          // 业务处理：同步渲染、统计丢包率
          console.log('Received Frame ID:', frameId);
        }
      } catch (e) { /* 扩展不存在或解析失败 */ }

      // 2. 读取 AI Tags
      try {
        const aiBuf = frame.getHeaderExtension(EXT_ID_AI_TAGS);
        if (aiBuf) {
          const tags = new TextDecoder().decode(aiBuf);
          // 触发 UI 更新或后续推理
          handleAiTags(JSON.parse(tags));
        }
      } catch (e) { /* ... */ }

      controller.enqueue(frame); // 透传给解码器
    }
  });

  readable.pipeThrough(processor).pipeTo(writable).catch(console.error);
}

sequenceDiagram
    participant App
    participant Encoder (WebCodecs/Insertable)
    participant TransformStream
    participant Network
    
    App->>Encoder: VideoFrame (Raw)
    Encoder-->>TransformStream: RTCEncodedVideoFrame (Encrypted Payload?)
    Note right of TransformStream: 1. 获取帧数据<br/>2. 派生帧密钥<br/>3. AES-GCM 加密 Payload<br/>4. 认证标签写入 Header Extension
    TransformStream->>Network: RTCEncodedVideoFrame (Ciphertext + Auth Tag)

// 发送端加密 Transform
async function encryptFrame(frame, keyManager) {
  // 1. 获取明文 Payload
  const plaintext = frame.data; 
  
  // 2. 派生 Nonce (通常基于 Frame Counter + Key ID)
  const nonce = keyManager.generateNonce(frame.timestamp);
  
  // 3. 加密 (Web Crypto API)
  const ciphertext = await crypto.subtle.encrypt(
    { name: 'AES-GCM', iv: nonce, tagLength: 128 },
    keyManager.getEncryptionKey(),
    plaintext
  );
  
  // 4. 构造新帧
  return new RTCEncodedVideoFrame({
    type: frame.type,
    timestamp: frame.timestamp,
    data: ciphertext, // 密文
    // 将 Auth Tag 或 Key ID 写入 Header Extension 供解密端使用
    // frame.setHeaderExtension(EXT_ID_SFRAME, authTagBuffer) 
  });
}

// 主线程
const worker = new Worker('processor.js');
readable.pipeTo(worker.writable); // 需配合 MessageChannel 或 Comlink 库实现流桥接

// audio-redundancy-processor.js
const EXT_ID_RED_FLAG = 13; // SDP: a=extmap:13 urn:ietf:params:rtp-hdrext:red-flag

class AudioRedundancyMarker {
  constructor(opusEncoder) {
    this.encoder = opusEncoder; // 假设可访问编码器配置或外部 FEC 模块
    this.frameCount = 0;
  }

  transform(frame, controller) {
    // 1. 音频帧极小，避免在 transform 中创建新对象/ArrayBuffer，直接修改引用传递
    // 注意：frame.data 为 ArrayBuffer，若需修改 Payload 必须新建；仅写 Header Extension 可原地操作
    
    // 2. 业务逻辑：每 3 帧发送一次 RED 冗余 (模拟策略)
    const isRedFrame = (this.frameCount++ % 3 === 0);
    
    if (isRedFrame) {
      // 标记 Header Extension: 1 byte (0x01 表示包含 RED)
      // 实际 RED 数据需在编码层 (WebCodecs AudioEncoder) 或 SFU 层生成并拼接到 Payload
      frame.setHeaderExtension(EXT_ID_RED_FLAG, new Uint8Array([1]).buffer);
    } else {
      // 显式清除或设为 0，防止旧内存残留被误读
      frame.setHeaderExtension(EXT_ID_RED_FLAG, new Uint8Array([0]).buffer);
    }

    controller.enqueue(frame);
  }
}

// mediasoup-router-setup.js
const router = await worker.createRouter({
  mediaCodecs: [ /* ... */ ],
  rtpHeaderExtensions: [
    // 标准扩展
    { uri: 'urn:ietf:params:rtp-hdrext:sdes:mid' },
    { uri: 'urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id' },
    { uri: 'urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id' },
    // 自定义扩展 (必须与客户端 SDP 完全一致)
    { uri: 'http://example.com/extension/frame-id', preferredId: 11, preferredEncrypt: false },
    { uri: 'http://example.com/extension/ai-tags', preferredId: 12, preferredEncrypt: false },
    { uri: 'urn:ietf:params:rtp-hdrext:red-flag', preferredId: 13, preferredEncrypt: false }, // 音频
  ]
});

// mediasoup Worker 侧逻辑伪代码 (C++/Rust 实现，JS 无法直接干预 Worker 数据平面)
// 实际生产建议编写 mediasoup C++ 扩展或使用支持脚本化的媒体服务器

class SmartPacketFilter {
  onRtpPacket(packet, producer) {
    // 1. 解析 Header Extension (需解析 RTP 头部 One-Byte/Two-Byte Header)
    const extMap = producer.rtpParameters.headerExtensions;
    const frameIdExt = extMap.find(e => e.uri === 'http://example.com/extension/frame-id');
    
    if (frameIdExt) {
      const frameId = packet.readHeaderExtension(frameIdExt.id); // 伪代码 API
      const priority = this.getBusinessPriority(frameId); // 查询业务 Redis/内存表
      
      // 2. 拥塞控制决策
      if (this.isCongested() && priority === 'LOW') {
        return PacketAction.DROP; // 直接丢包，不转发给 Consumer
      }
    }
    return PacketAction.FORWARD;
  }
}

// extensions.proto
syntax = "proto3";
package rtc.extensions;

message FrameMetadata {
  uint32 frame_id = 1;
  int64 server_timestamp_ms = 2;
  // 视频专用
  VideoLayerInfo video_layer = 3;
  // 音频专用
  AudioRedundancyInfo audio_red = 4;
  // 通用 AI 标签
  repeated AiTag ai_tags = 5;
}

message VideoLayerInfo {
  uint32 spatial_idx = 1;
  uint32 temporal_idx = 2;
  bool is_key_frame = 3;
}

# .github/workflows/webrtc-e2e.yml
jobs:
  insertable-streams-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Start Headless Chrome + Fake Media
        run: |
          # 使用 chrome-headless-shell + --use-fake-device-for-media-stream
          # 注入合成视频轨 (YUV420P) 和音频轨 (Sine Wave)
      - name: Run Mocha Tests
        run: npm test
      - name: Validate RTP Extensions
        run: |
          # 1. 解析测试产生的 .pcapng 文件
          # 2. 用 tshark 提取所有 RTP 包的 Header Extension
          # 3. 断言：Frame ID 单调递增、AI Tags 格式合法、RED Flag 周期性出现
          python scripts/validate_rtp_extensions.py artifacts/dump.pcapng

// media-pipeline-factory.ts
enum PipelineMode { LEGACY_DATACHANNEL, INSERTABLE_STREAMS, HYBRID }

class MediaPipelineFactory {
  static createSender(pc: RTCPeerConnection, track: MediaStreamTrack, config: AppConfig) {
    // 1. 能力探测
    const supportInsertable = 'createEncodedStreams' in RTCRtpSender.prototype;
    
    // 2. 灰度策略 (配置中心下发)
    const mode = config.getFlag('webrtc_insertable_rollout') 
      ? (supportInsertable ? PipelineMode.INSERTABLE_STREAMS : PipelineMode.LEGACY_DATACHANNEL)
      : PipelineMode.LEGACY_DATACHANNEL;

    switch (mode) {
      case PipelineMode.INSERTABLE_STREAMS:
        return new InsertableStreamSender(pc, track);
      case PipelineMode.HYBRID:
        // 关键帧走 Insertable (低延迟)，元数据走 DataChannel (兼容旧版 SFU)
        return new HybridSender(pc, track);
      default:
        return new LegacyDataChannelSender(pc, track);
    }
  }
}

// rtc-transform-adapter.ts - 屏蔽版本差异
export interface IFrameProcessor {
  process(frame: EncodedFrameView): EncodedFrameView | Promise<EncodedFrameView>;
}

// 适配当前 Insertable Streams
class CurrentTransformAdapter implements IFrameProcessor {
  constructor(private transformer: TransformStreamDefaultController<RTCEncodedVideoFrame>) {}
  
  async process(frame: RTCEncodedVideoFrame) {
    // 当前逻辑...
    this.transformer.enqueue(newFrame);
  }
}

// 适配未来 RTCRtpScriptTransform (Worker Context)
class FutureWorkerAdapter implements IFrameProcessor {
  // 实现标准 transform(controller) 签名
  async transform(controller, frame) {
    const processed = await this.coreLogic(frame);
    controller.enqueue(processed);
  }
}

// 业务核心逻辑完全复用
class BusinessLogic {
  static async coreLogic(frame: EncodedFrameView) {
    // 注入元数据、加密、AI 标记... 纯函数，无平台依赖
    return frame;
  }
}

// 启用
const pc = new RTCPeerConnection({ encodedInsertableStreams: true });

// 发送端
const { readable, writable } = sender.createEncodedStreams();
readable.pipeThrough(new TransformStream({
  transform(frame, ctrl) {
    // frame: RTCEncodedVideoFrame | RTCEncodedAudioFrame
    // frame.data: ArrayBuffer (Payload)
    // frame.timestamp: number (RTP Timestamp)
    // frame.type: 'key' | 'delta' | 'empty'
    // frame.getHeaderExtension(id): ArrayBuffer | null
    // frame.setHeaderExtension(id, buffer): void
    // frame.metadata: any (管道内传递，不上网)
    ctrl.enqueue(modifiedFrame);
  }
})).pipeTo(writable);

// 接收端
const { readable, writable } = receiver.createEncodedStreams();
// 同理 pipeThrough 处理

# 视频扩展
a=extmap:10 urn:ietf:params:rtp-hdrext:toffset
a=extmap:11 http://your.domain/ext/frame-id
a=extmap:12 http://your.domain/ext/ai-tags
a=extmap:13 http://your.domain/ext/video-layer

# 音频扩展
a=extmap:14 urn:ietf:params:rtp-hdrext:sdes:mid
a=extmap:15 http://your.domain/ext/audio-red
a=extmap:16 http://your.domain/ext/audio-vad