graph LR
    A[发送端] -->|1. 信令交换| B(信令服务器)
    C[接收端] -->|1. 信令交换| B
    B -->|2. SDP/ICE Candidate| A
    B -->|2. SDP/ICE Candidate| C
    A -->|3. 建立 DataChannel| C
    A -->|4. 文件分块/校验/传输| C
    C -->|5. 进度反馈/断点记录| A

// file-chunker.worker.js
self.onmessage = async ({ data: { file, chunkSize = 1024 * 1024 } }) => {
  const chunks = [];
  const spark = new SparkMD5.ArrayBuffer();
  let offset = 0;
  let index = 0;

  while (offset < file.size) {
    const slice = file.slice(offset, offset + chunkSize);
    const arrayBuffer = await slice.arrayBuffer();
    spark.append(arrayBuffer);
    chunks.push({
      index,
      start: offset,
      end: offset + slice.size,
      size: slice.size,
      hash: null // 可选：单片哈希，便于细粒度校验
    });
    offset += slice.size;
    index++;
    // 进度回报主线程
    if (index % 50 === 0) self.postMessage({ type: 'progress', percent: offset / file.size });
  }

  self.postMessage({
    type: 'complete',
    payload: {
      chunks,
      totalChunks: index,
      fileHash: spark.end(), // 整文件 SHA-256/MD5
      fileName: file.name,
      fileSize: file.size,
      mimeType: file.type
    }
  });
};

// rtc-transport.js
class RTCFileTransport {
  constructor(signaling, config = {}) {
    this.pc = new RTCPeerConnection(config.iceServers);
    this.channel = this.pc.createDataChannel('file-transfer', {
      ordered: true,        // 保证顺序，配合滑动窗口实现可靠性
      maxRetransmits: 0,    // 0 表示无限重传（可靠模式）
      protocol: 'file.v1'   // 子协议标识，便于扩展
    });
    this._bindEvents();
    this.signaling = signaling;
  }

  _bindEvents() {
    this.channel.onopen = () => this._onOpen();
    this.channel.onclose = () => this._onClose();
    this.channel.onerror = (e) => this._onError(e);
    this.channel.onmessage = (e) => this._onMessage(e.data);
    this.pc.onicecandidate = (e) => e.candidate && this.signaling.send({ type: 'candidate', candidate: e.candidate });
  }

  async start(isInitiator, remoteSdp) {
    if (isInitiator) {
      const offer = await this.pc.createOffer();
      await this.pc.setLocalDescription(offer);
      this.signaling.send({ type: 'offer', sdp: offer });
    } else {
      await this.pc.setRemoteDescription(remoteSdp);
      const answer = await this.pc.createAnswer();
      await this.pc.setLocalDescription(answer);
      this.signaling.send({ type: 'answer', sdp: answer });
    }
  }
}

// flow-control.js
const HIGH_WATER_MARK = 16 * 1024 * 1024; // 16MB
const LOW_WATER_MARK  = 4 * 1024 * 1024;  // 4MB

class SlidingWindowSender {
  constructor(channel, chunks, windowSize = 32) {
    this.channel = channel;
    this.chunks = chunks;
    this.windowSize = windowSize;
    this.base = 0;           // 窗口左边界
    this.next = 0;           // 下一个待发送分片索引
    this.pending = new Map(); // index -> { data, retries, timer }
    this._setupBackpressure();
  }

  _setupBackpressure() {
    this.channel.bufferedAmountLowThreshold = LOW_WATER_MARK;
    this.channel.onbufferedamountlow = () => this._flushWindow();
  }

  _flushWindow() {
    while (this.next < this.base + this.windowSize && this.next < this.chunks.length) {
      if (this.channel.bufferedAmount > HIGH_WATER_MARK) break; // 背压触发
      this._sendChunk(this.next++);
    }
  }

  _sendChunk(index) {
    const chunkMeta = this.chunks[index];
    const fileSlice = this.file.slice(chunkMeta.start, chunkMeta.end);
    const packet = this._encodePacket(index, fileSlice);
    this.channel.send(packet);
    this.pending.set(index, { data: packet, retries: 0, timer: setTimeout(() => this._retransmit(index), 5000) });
  }

  _onAck(index) {
    clearTimeout(this.pending.get(index)?.timer);
    this.pending.delete(index);
    this.base = Math.max(this.base, index + 1);
    this._persistProgress(this.base); // 断点记录
    this._flushWindow();
  }
}

// persistence.js
const DB_NAME = 'FileTransferDB';
const STORE_NAME = 'sessions';

function openDB() {
  return new Promise((resolve, reject) => {
    const req = indexedDB.open(DB_NAME, 1);
    req.onupgradeneeded = (e) => {
      const db = e.target.result;
      if (!db.objectStoreNames.contains(STORE_NAME)) {
        db.createObjectStore(STORE_NAME, { keyPath: 'sessionId' });
      }
    };
    req.onsuccess = () => resolve(req.result);
    req.onerror = () => reject(req.error);
  });
}

async function saveSession(session) {
  const db = await openDB();
  return new Promise((resolve, reject) => {
    const tx = db.transaction(STORE_NAME, 'readwrite');
    tx.objectStore(STORE_NAME).put(session);
    tx.oncomplete = resolve;
    tx.onerror = () => reject(tx.error);
  });
}

async function loadSession(sessionId) {
  const db = await openDB();
  return new Promise((resolve, reject) => {
    const req = db.transaction(STORE_NAME).objectStore(STORE_NAME).get(sessionId);
    req.onsuccess = () => resolve(req.result);
    req.onerror = () => reject(req.error);
  });
}

// 会话数据结构示例
const session = {
  sessionId: 'uuid-v4',
  fileMeta: { name: 'demo.zip', size: 2_147_483_648, hash: 'sha256...' },
  chunks: [/* chunkMeta[] */],
  transferredChunks: new Uint8Array(Math.ceil(totalChunks / 8)), // 位图
  lastActive: Date.now(),
  remotePeerId: 'peer-xxx'
};

graph TB
    Client[客户端集群] -->|WebSocket/HTTP3| LB[负载均衡层<br/>Nginx/Envoy/HAProxy]
    LB -->|一致性哈希/Sticky Session| SignalCluster[信令集群<br/>Stateless Node.js/Go/Rust]
    SignalCluster -->|Pub/Sub| RedisCluster[(Redis Cluster<br/>房间/在线状态/消息路由)]
    SignalCluster -->|持久化| PG[(PostgreSQL<br/>审计日志/计费记录)]
    SignalCluster -->|指标上报| Prometheus[Prometheus + Grafana]
    TURN[TURN/STUN 集群] -.->|ICE 候选下发| SignalCluster

# docker-compose.turn.yml
services:
  turn:
    image: coturn/coturn:latest
    network_mode: host
    command: >
      -n --no-cli
      --realm=turn.example.com
      --lt-cred-mech
      --user=username:password
      --min-port=49152 --max-port=65535
      --external-ip=203.0.113.10/10.0.0.5  # 公网IP/内网IP 映射
      --tls-listening-port=443
      --alt-tls-listening-port=8443
      --cert=/etc/turn/cert.pem --pkey=/etc/turn/key.pem
      --log-file=stdout
      --no-multicast-peers
      --max-bps=100000000  # 单会话限速 100Mbps 防滥用
    volumes:
      - ./certs:/etc/turn
    deploy:
      replicas: 3
      placement:
        constraints: [node.role == manager]
      restart_policy:
        condition: on-failure

// ice-restart-strategy.js
class ICEManager {
  constructor(pc, signaling) {
    this.pc = pc;
    this.signaling = signaling;
    this.failedCount = 0;
    this._bindEvents();
  }

  _bindEvents() {
    this.pc.oniceconnectionstatechange = () => this._onStateChange();
    this.pc.onconnectionstatechange = () => this._onConnStateChange();
  }

  _onStateChange() {
    const state = this.pc.iceConnectionState;
    if (['disconnected', 'failed'].includes(state)) {
      this.failedCount++;
      if (this.failedCount <= 3) {
        // 轻度异常：触发 ICE Restart（保留 DataChannel、媒体轨道）
        this._triggerRestart('ice-restart');
      } else {
        // 重度异常：全量重连（重建 PeerConnection）
        this._triggerRestart('full-reconnect');
      }
    } else if (state === 'connected') {
      this.failedCount = 0; // 重置计数器
    }
  }

  async _triggerRestart(type) {
    if (type === 'ice-restart') {
      const offer = await this.pc.createOffer({ iceRestart: true });
      await this.pc.setLocalDescription(offer);
      this.signaling.send({ type: 'offer', sdp: offer, restart: true });
    } else {
      this.signaling.send({ type: 'reconnect-request', sessionId: this.sessionId });
      // 上层应用层处理重建逻辑
    }
  }
}

// protocol.ts
enum ControlType {
  HANDSHAKE = 0x01,      // 握手：协商版本、窗口大小、压缩算法
  FILE_META = 0x02,      // 文件元数据：name, size, hash, chunks[]
  CHUNK_DATA = 0x03,     // 普通数据分片
  ACK = 0x04,            // 确认：ackSeqNum, sackBlocks[] (SACK)
  NACK = 0x05,           // 否定确认：missingSeqNums[]
  RESUME = 0x06,         // 断点续传：bitmap, startOffset
  HEARTBEAT = 0x07,      // 心跳：timestamp
  ERROR = 0xFF,          // 错误：code, message
}

// 示例：RESUME 帧 Payload 编码
function encodeResumeFrame(bitmap: Uint8Array, startOffset: number): ArrayBuffer {
  const header = new DataView(new ArrayBuffer(12));
  header.setUint16(0, 0x4654, true);
  header.setUint8(2, 1);
  header.setUint8(3, 0);
  header.setUint16(4, streamId, true);
  header.setUint32(6, seqNum++, true);
  
  const payload = new Uint8Array(1 + 8 + bitmap.length); // type(1) + offset(8) + bitmap
  payload[0] = ControlType.RESUME;
  new DataView(payload.buffer).setBigUint64(1, BigInt(startOffset), true);
  payload.set(bitmap, 9);
  
  header.setUint32(10, payload.length, true);
  return concatBuffers(header.buffer, payload.buffer);
}

// main/preload.ts (Main Process)
const { ipcMain } = require('electron');
const crypto = require('node:crypto');
const { Readable } = require('node:stream');

ipcMain.handle('file:hash-chunks', async (_, filePath, chunkSize = 1024 * 1024) => {
  const hash = crypto.createHash('sha256');
  const chunks = [];
  let offset = 0;
  let index = 0;

  for await (const chunk of Readable.toWeb(fs.createReadStream(filePath, { highWaterMark: chunkSize }))) {
    const buf = Buffer.from(await chunk.arrayBuffer());
    hash.update(buf);
    chunks.push({ index, start: offset, end: offset + buf.length, size: buf.length });
    offset += buf.length;
    index++;
  }
  return { chunks, fileHash: hash.digest('hex'), totalChunks: index };
});

// 客户端发起传输时生成
const traceId = crypto.randomUUID();
const spanId = crypto.randomBytes(8).toString('hex');

// 1. 信令消息携带
signaling.send({ type: 'offer', sdp, traceId, spanId });

// 2. DataChannel 首帧 (HANDSHAKE) 携带
channel.send(encodeHandshake({ traceId, parentSpanId: spanId }));

// 3. 服务端/接收端收到后，提取 traceId 关联日志、指标、Jaeger/Zipkin Span
// Jaeger Span 示例
const span = tracer.startSpan('webrtc-file-transfer', {
  childOf: tracer.extract(FORMAT_TEXT_MAP, { 'traceparent': `00-${traceId}-${spanId}-01` }),
  tags: { 'file.size': fileSize, 'peer.id': remotePeerId }
});

// 监听 visibilitychange，主动探活
document.addEventListener('visibilitychange', async () => {
  if (document.visibilityState === 'visible' && dc?.readyState === 'open') {
    try {
      // 发送一个 0 字节探测帧，触发 DTLS 重协商或暴露真实错误
      dc.send(new ArrayBuffer(0)); 
    } catch (e) {
      // 捕获 InvalidStateError 等，触发 ICE Restart 或重连
      await handleTransportFailure();
    }
  }
});

// load-test.js (k6)
import ws from 'k6/ws';
import { check, sleep } from 'k6';
import { SharedArray } from 'k6/data';

// 预生成 100MB 随机二进制数据 (内存常驻)
const testData = new SharedArray('100MB-payload', () => {
  const arr = new Uint8Array(100 * 1024 * 1024);
  crypto.getRandomValues(arr);
  return arr.buffer;
});

export const options = {
  stages: [
    { duration: '2m', target: 50 },   // 预热
    { duration: '10m', target: 200 }, // 压力平台期
    { duration: '2m', target: 0 },    // 降压
  ],
  thresholds: {
    'ws_connecting': ['avg<2000'],
    'ws_session_duration': ['p(95)<300000'], // 5分钟内完成
    'checks': ['rate>0.99'],
  },
};

export default function () {
  const url = `wss://signal.example.com/ws?token=${__VU}`;
  const res = ws.connect(url, {}, function (socket) {
    socket.on('open', () => {
      // 1. 信令交换 (简化)
      socket.send(JSON.stringify({ type: 'join', room: 'load-test' }));
    });

    socket.on('message', (msg) => {
      const data = JSON.parse(msg);
      if (data.type === 'offer') {
        // 2. 模拟建立 DataChannel 并发送数据 (此处需集成真实 WebRTC 库或 Mock)
        // k6 当前不支持原生 WebRTC，生产压测建议使用 **自定义 Go/Node 客户端集群**
        // 此处仅演示指标上报逻辑
        const start = Date.now();
        // 模拟分片发送耗时
        sleep(Math.random() * 10 + 5); 
        const duration = Date.now() - start;
        // 上报自定义指标
        // (需配合 xk6-output-prometheus-remote 或 statsd)
      }
    });

    socket.on('close', () => {});
  });
  check(res, { 'status 101': (r) => r && r.status === 101 });
}

# 使用 pnpm + Turborepo
npx create-turbo@latest webrtc-file-transfer
cd webrtc-file-transfer

# Apps
# apps/web-sdk       -> 前端 SDK (TypeScript, Vite, Vitest)
# apps/signaling     -> 信令网关 (Go, Hertz, Wire, Prometheus)
# apps/turn          -> TURN 部署 (Dockerfile, Helm Chart)
# apps/load-test     -> 压测客户端 (Go, pion/webrtc)

# Packages
# packages/protocol  -> 共享协议定义 (Protobuf/TypeScript 双端生成)
# packages/crypto    -> 共享加密工具 (Web Crypto / Go crypto 统一接口)
# packages/logger    -> 统一日志规范 (Zap / Pino / Console 适配)

pnpm install
turbo run build test lint