graph LR
    A[Client A<br/>Publisher] -->|WebTransport<br/>Datagram/Stream| B(Server<br/>Go/Rust Node)
    C[Client B<br/>Subscriber] -->|WebTransport<br/>Datagram/Stream| B
    B -->|转发媒体流| C
    B -->|转发信令/控制帧| A

package main

import (
    "context"
    "crypto/tls"
    "log"
    "net/http"
    "golang.org/x/net/http3"
    "golang.org/x/net/webtransport"
)

func main() {
    // 1. TLS 配置 (需支持 ALPN h3)
    tlsConfig := &tls.Config{
        Certificates: []tls.Certificate{loadCert("cert.pem", "key.pem")},
        NextProtos:   []string{"h3", "h3-29"}, // HTTP/3 ALPN
    }

    // 2. WebTransport 升级处理器
    wtServer := &webtransport.Server{
        CheckOrigin: func(r *http.Request) bool { return true }, // 生产环境需严格校验 Origin
    }

    http.HandleFunc("/wt", func(w http.ResponseWriter, r *http.Request) {
        // 升级为 WebTransport 连接
        conn, err := wtServer.Upgrade(w, r)
        if err != nil {
            log.Printf("Upgrade failed: %v", err)
            return
        }
        go handleSession(context.Background(), conn)
    })

    // 3. 启动 HTTP/3 服务
    server := &http3.Server{
        Addr:      ":4433",
        TLSConfig: tlsConfig,
        Handler:   http.DefaultServeMux,
    }
    log.Fatal(server.ListenAndServe())
}

func handleSession(ctx context.Context, session *webtransport.Session) {
    defer session.CloseWithError(0, "closed")
    
    // 接受单向流 (媒体数据) 或双向流 (信令)
    for {
        stream, err := session.AcceptStream(ctx)
        if err != nil { break }
        go handleStream(stream)
    }
}

// client-transport.js
class WTClient {
  constructor(url) {
    this.transport = new WebTransport(url);
    this.datagramQueue = []; // 发送缓冲队列
    this.streams = new Map(); // streamId -> ReadableStreamDefaultReader
  }

  async connect() {
    try {
      await this.transport.ready;
      console.log('WebTransport 连接已就绪');
      this.setupIncomingStreams();
      this.startDatagramLoop();
    } catch (err) {
      console.error('连接失败:', err);
      // 降级策略：尝试 WebRTC 或 WebSocket
    }
  }

  // 处理服务端主动推流 (单向流)
  async setupIncomingStreams() {
    for await (const stream of this.transport.incomingUnidirectionalStreams) {
      const reader = stream.getReader();
      this.processMediaStream(reader); // 解复用媒体包
    }
  }
}

async function sendVideoFrame(frameData, timestamp, isKeyFrame) {
  const MTU = 1200; // 经验值，留出 QUIC/UDP/IP 头部空间
  const headerSize = 13;
  const maxPayload = MTU - headerSize;
  const fragments = Math.ceil(frameData.byteLength / maxPayload);
  const frameId = ++this.frameIdCounter;

  for (let i = 0; i < fragments; i++) {
    const start = i * maxPayload;
    const end = Math.min(start + maxPayload, frameData.byteLength);
    const payload = frameData.slice(start, end);

    // 构造二进制头部
    const header = new ArrayBuffer(13);
    const view = new DataView(header);
    view.setUint32(0, frameId, true);
    view.setUint16(4, i, true);
    view.setUint16(6, fragments, true);
    view.setUint8(8, isKeyFrame ? 0 : 1);
    view.setUint32(9, timestamp, true);

    // 合并发送
    const packet = new Blob([header, payload]);
    
    // 关键：使用 sendDatagram (不可靠，低延迟)
    // 若队列满需背压处理
    try {
      await this.transport.datagrams.send(packet);
    } catch (e) {
      // Datagram 发送失败通常意味着拥塞或连接关闭
      this.handleCongestionSignal();
    }
  }
}

// 简易重组缓冲区
const reassemblyBuffer = new Map(); // frameId -> { fragments: Map<index, data>, count, timer }

function processDatagram(dataView) {
  const frameId = dataView.getUint32(0, true);
  const fragIndex = dataView.getUint16(4, true);
  const fragCount = dataView.getUint16(6, true);
  // ... 解析头部 ...

  if (!reassemblyBuffer.has(frameId)) {
    reassemblyBuffer.set(frameId, { fragments: new Map(), count: fragCount, timer: setTimeout(() => cleanup(frameId), 100) });
  }
  const frame = reassemblyBuffer.get(frameId);
  frame.fragments.set(fragIndex, dataView.buffer.slice(13)); // 存载荷

  // 判断是否收齐
  if (frame.fragments.size === frame.count) {
    clearTimeout(frame.timer);
    const fullFrame = concatFragments(frame.fragments, frame.count);
    decodeAndRender(fullFrame); // 送入解码器
    reassemblyBuffer.delete(frameId);
  }
}

// Client -> Server: Join Room
{ "type": "join", "room": "meeting_101", "role": "publisher", "codecs": ["vp9", "opus"] }

// Server -> Client: Peer Joined / Track Info
{ "type": "peer_join", "peerId": "user_B", "tracks": [{ "trackId": "v1", "kind": "video", "simulcast": true }] }

// Client -> Server: Keyframe Request (PLI)
{ "type": "pli", "trackId": "v1", "senderId": "user_A" }

if ('WebTransport' in window && await checkDatagramSupport()) {
  initWebTransport();
} else {
  initWebRTCFallback(); // 复用现有 SFU 架构
}

graph TB
    subgraph Main Thread
        A[Video Element / Canvas] -->|requestVideoFrameCallback| B(VideoFrame)
    end
    
    subgraph Encoding Worker (OffscreenCanvas / Dedicated Worker)
        B -->|postMessage transfer| C[VideoEncoder.encode]
        C -->|EncodedVideoChunk| D[分片打包模块]
        D -->|Datagram| E[WebTransport Send]
    end
    
    subgraph Decoding Worker
        F[WebTransport Recv] -->|重组| G[Decoder.decode]
        G -->|VideoFrame| H[VideoFrame.copyTo / drawImage]
        H --> I[OffscreenCanvas / WebGL Texture]
    end

// encoding-worker.js
let encoder = null;
let currentBitrate = 2_000_000; // 2Mbps 初始值
const TARGET_FPS = 30;

async function initEncoder(codec = 'avc1.42001f') { // Baseline Profile 兼容性最好
  encoder = new VideoEncoder({
    output: handleEncodedChunk,
    error: (e) => console.error('Encoder Error:', e)
  });

  encoder.configure({
    codec,
    width: 1280,
    height: 720,
    bitrate: currentBitrate,
    framerate: TARGET_FPS,
    hardwareAcceleration: 'prefer-hardware', // 关键：优先硬编
    latencyMode: 'realtime',                 // 关键：实时模式，禁用 B 帧/前向参考
  });
}

// 处理编码输出：直接发送 Datagram，避免中间缓冲
function handleEncodedChunk(chunk, metadata) {
  // metadata.decoderConfigDescription 可用于信令交换 SPS/PPS
  if (chunk.type === 'key') {
    // 关键帧携带配置信息，或通过信令单独下发
    sendConfigIfNeeded(metadata.decoderConfigDescription);
  }
  
  // 直接分片发送 (复用上篇分片逻辑)
  // 注意：chunk.data 是 ArrayBuffer，零拷贝传入分片器
  fragmentAndSend(chunk.data, chunk.timestamp, chunk.type === 'key');
}

// 动态码率调整接口 (由拥塞控制模块调用)
function updateBitrate(newBitrate) {
  if (Math.abs(newBitrate - currentBitrate) / currentBitrate > 0.1) { // 变化>10%才重配
    currentBitrate = newBitrate;
    encoder.configure({ bitrate: currentBitrate }); // 运行时重配，无需销毁重建
  }
}

// 请求关键帧 (PLI 响应)
function requestKeyFrame() {
  encoder.encode({ keyFrame: true }); // 标准 API 强制 IDR
}

// VP9 SVC 3层配置示例 (L1: 180p/15fps, L2: 360p/30fps, L3: 720p/30fps)
const svcConfig = {
  codec: 'vp09.00.10.08', // VP9 Profile 0
  scalabilityMode: 'L3T3_KEY', // 3层空间 + 3层时间 (标准模式)
  // 也可自定义: 'L3T1_KEY' (仅空间分层，固定帧率)
};

encoder.configure({
  ...baseConfig,
  ...svcConfig,
  bitrate: 2_500_000, // 总目标码率
  // 可选：显式指定各层码率分配 (需编码器支持)
  // scalabilityMode: 'L3T3', 
  // bitrateAllocation: [
  //   { bitrate: 150_000, framerate: 15, resolution: [320, 180] }, // L1
  //   { bitrate: 600_000, framerate: 30, resolution: [640, 360] }, // L2
  //   { bitrate: 1_750_000, framerate: 30, resolution: [1280, 720] } // L3
// ]);

// internal/sfu/forwarder.go
type Subscriber struct {
    ID           string
    Session      *webtransport.Session
    MaxLayer     int       // 订阅的最高空间层 (0,1,2)
    MaxTemporal  int       // 订阅的最高时间层
    SendChan     chan *wirePacket // 发送协程专用通道，串行化写入
}

type Room struct {
    sync.RWMutex
    Publisher    *Publisher
    Subscribers  map[string]*Subscriber
}

// 发布者接收协程 -> 分发给订阅者
func (r *Room) DispatchLoop(ctx context.Context) {
    for {
        select {
        case pkt := <-r.Publisher.IncomingPackets:
            // 1. 解析 VP9/AV1 Header 获取 SID/TID (零拷贝解析)
            sid, tid := parseLayerIDs(pkt.Payload)
            
            // 2. 扇出给符合条件的订阅者 (非阻塞发送)
            r.RLock()
            for _, sub := range r.Subscribers {
                if sid <= sub.MaxLayer && tid <= sub.MaxTemporal {
                    select {
                    case sub.SendChan <- pkt: // 通道满即丢包，触发背压
                    default:
                        sub.Stats.DroppedPackets++
                    }
                }
            }
            r.RUnlock()
        case <-ctx.Done():
            return
        }
    }
}

// 订阅者发送协程 (每订阅者一协程，保证发送序)
func (s *Subscriber) SendLoop(ctx context.Context) {
    stream, _ := s.Session.OpenUniStream() // 复用单向流发媒体
    writer := stream.Writer()
    defer writer.Close()
    
    for {
        select {
        case pkt := <-s.SendChan:
            // 直接写入 QUIC Stream (可靠) 或 Datagram (不可靠)
            // 建议：关键帧/配置帧走 Stream，增强层走 Datagram
            if pkt.IsKeyFrame || pkt.IsConfig {
                writer.Write(pkt.Bytes) // 可靠流
            } else {
                s.Session.SendDatagram(pkt.Bytes) // 不可靠
            }
        case <-ctx.Done():
            return
        }
    }
}

// crypto/sframe.js (简化版，生产建议用 @wireapp/sframe 或 libsodium.wasm)
import { AES_GCM } from './aes-gcm-wrapper'; // Web Crypto API 封装

class SFrameContext {
  constructor(keyId, baseKey) {
    this.keyId = keyId; // Key ID (KID)
    this.baseKey = baseKey; // CryptoKey (AES-GCM)
    this.counter = new Uint8Array(8); // 64-bit 计数器
  }

  // 加密: 输入明文帧 -> 输出 SFrame 密文
  async encrypt(plaintext, metadata) {
    // 1. 构造 Nonce: KID (1B) + Counter (8B) 
    const nonce = new Uint8Array(9);
    nonce[0] = this.keyId;
    nonce.set(this.counter, 1);
    this.incrementCounter();

    // 2. AAD: 可选元数据 (如帧类型、时间戳哈希) 防重排
    const aad = this.buildAAD(metadata);

    // 3. AES-GCM 加密
    const ciphertext = await AES_GCM.encrypt(this.baseKey, plaintext, nonce, aad);
    
    // 4. 组装 SFrame 头部: KID (VLQ) + CTR (VLQ) + Ciphertext + Tag
    return this.buildSFrameHeader(this.keyId, this.counter) + ciphertext;
  }

  // 解密: 验证 Tag -> 输出明文
  async decrypt(sframeCiphertext) {
    const { kid, ctr, ciphertext, tag } = this.parseSFrameHeader(sframeCiphertext);
    // 查找对应 KID 的 Key (支持密钥轮换过渡期双密钥)
    const key = this.getKey(kid); 
    const nonce = buildNonce(kid, ctr);
    return AES_GCM.decrypt(key, ciphertext, nonce, this.buildAAD({}));
  }
}

// scheduler/priority_queue.go
type PacketPriority int
const (
    PrioKeyFrame PacketPriority = iota // 最高
    PrioFECKey
    PrioAudio
    PrioDeltaFrame
    PrioFECDelta
    PrioPadding // 填充包，探测带宽
)

type Scheduler struct {
    queues [6]chan *Packet
    budget *TokenBucket // 令牌桶，速率由 GCC 控制
}

func (s *Scheduler) Run(ctx context.Context, sendFn func(*Packet)) {
    for {
        select {
        case <-ctx.Done(): return
        default:
            // 严格优先级抢占调度
            var pkt *Packet
            for p := PrioKeyFrame; p <= PrioPadding; p++ {
                select {
                case pkt = <-s.queues[p]:
                    goto SEND
                default:
                }
            }
            // 无包可发，短暂休眠
            time.Sleep(100 * time.Microsecond)
            continue

        SEND:
            // 预算检查
            if s.budget.Take(pkt.Size) {
                sendFn(pkt)
            } else {
                // 预算不足，放回队列头 (或丢弃低优先级)
                s.requeue(pkt, p)
                time.Sleep(time.Millisecond)
            }
        }
    }
}

// 核心指标采集循环 (每 2s 上报一次)
setInterval(async () => {
  const wtStats = await transport.getStats(); // WebTransportStat
  const codecStats = videoDecoder.getStats(); // VideoDecoderStat
  
  const payload = {
    // 传输层
    rtt: wtStats.rtt,
    packetsLost: wtStats.datagramsLost,
    sendBitrate: wtStats.bytesSent * 8 / 2, // bps
    
    // 解码层
    decodeTime: codecStats.decodeTime, // 平均解码耗时 ms
    framesDropped: codecStats.framesDropped,
    frameDelay: performance.now() - lastFrameTimestamp, // 端到端延迟估算
    
    // 业务层
    currentLayer: currentSpatialLayer,
    freezeDuration: accumulator.freezeTime, // 卡顿累计时长
  };
  
  // 批量上报至 ClickHouse / VictoriaMetrics
  navigator.sendBeacon('/api/v1/metrics', JSON.stringify(payload));
}, 2000);

# 单用户连续 10s 丢包率 > 15% 且 码率 < 300kbps
avg_over_time(client_packet_loss_rate[10s]) > 0.15 
  and avg_over_time(client_send_bitrate[10s]) < 300000

# .gitlab-ci.yml / GitHub Actions
integration-test:
  services:
    - name: alpine:latest
      alias: netem
      command: ["/bin/sh", "-c", "tc qdisc add dev eth0 root netem loss 10% delay 100ms 20ms rate 1mbit && sleep infinity"]
  script:
    - docker run --cap-add=NET_ADMIN --network container:netem my-test-suite

Client <--WS/WSS--> Signaling Gateway (Stateless) <--gRPC--> Room Manager (Stateful)
                                                      |
                                                      +--> SFU Pool (Stateless)