graph LR
    A[发送端 Sender] -->|1. 文件分片 & 索引构建| B[Web Worker 分片引擎]
    B -->|2. 生成元数据 Manifest| C[Signal Server 信令交换]
    C -->|3. ICE Candidate & SDP| D[接收端 Receiver]
    D -->|4. DataChannel 建连| E[可靠传输层 Reliable Layer]
    E -->|5. 滑动窗口发送| F[SCTP 流控]
    F -->|6. ACK/NACK 反馈| E
    E -->|7. 进度持久化 IndexedDB| G[断点续传状态机]
    G -->|8. 重连恢复| A

// types.ts
export interface FileManifest {
  fileId: string;           // UUID v4
  fileName: string;
  fileSize: number;         // 字节
  chunkSize: number;        // 建议 16KB~256KB，适配MTU
  totalChunks: number;
  chunks: ChunkMeta[];      // 仅首包发送，后续增量同步
  createdAt: number;
  sha256: string;           // 全文件哈希，完成后校验
}

export interface ChunkMeta {
  index: number;            // 0-based
  offset: number;
  size: number;
  sha256: string;           // 分片哈希，支持并行校验
}

// reliable-channel.ts
const WINDOW_SIZE = 128;           // 并发在途分片数
const RETRANSMIT_TIMEOUT = 800;    // ms，动态RTT估算后调整
const MAX_RETRIES = 5;

interface Packet {
  seq: number;          // 全局单调递增
  chunkIndex: number;   // 所属分片索引
  offset: number;       // 分片内偏移
  data: ArrayBuffer;    // 实际载荷
  isLast: boolean;      // 分片结束标志
}

export class ReliableDataChannel {
  private dc: RTCDataChannel;
  private sendBase = 0;           // 窗口左边界
  private nextSeq = 0;            // 下一个分配序列号
  private ackMap = new Map<number, Packet>(); // 待确认包缓存
  private timers = new Map<number, NodeJS.Timeout>();
  private receiverWindow = 65535; // 对端通告窗口（字节）

  constructor(dc: RTCDataChannel) {
    this.dc = dc;
    this.dc.binaryType = 'arraybuffer';
    this.dc.onmessage = this.handleMessage.bind(this);
    this.dc.onbufferedamountlow = this.drainSendBuffer.bind(this);
  }

  // 发送端：分片入队
  async sendChunk(chunkIndex: number, data: ArrayBuffer): Promise<void> {
    const packets = this.fragment(chunkIndex, data);
    for (const pkt of packets) {
      await this.enqueuePacket(pkt);
    }
  }

  private async enqueuePacket(pkt: Packet): Promise<void> {
    // 流控：等待窗口可用
    while (this.nextSeq - this.sendBase >= WINDOW_SIZE) {
      await this.waitForAck();
    }
    // 缓存并发送
    this.ackMap.set(pkt.seq, pkt);
    this.startTimer(pkt.seq);
    this.dc.send(this.encode(pkt));
    this.nextSeq++;
  }

  // 接收端：ACK/NACK 生成
  private handleMessage(event: MessageEvent): void {
    const msg = this.decode(event.data);
    if (msg.type === 'DATA') {
      this.receiveBuffer.put(msg.seq, msg);
      this.sendAck(this.receiveBuffer.getContiguousEnd());
    } else if (msg.type === 'ACK') {
      this.processAck(msg.ackSeq, msg.sackBlocks);
    }
  }

  private processAck(ackSeq: number, sackBlocks: [number, number][]): void {
    // 1. 累积ACK：清理 sendBase 到 ackSeq
    for (let seq = this.sendBase; seq < ackSeq; seq++) {
      this.clearTimer(seq);
      this.ackMap.delete(seq);
    }
    this.sendBase = ackSeq;

    // 2. SACK块：快速重传触发
    for (const [start, end] of sackBlocks) {
      for (let seq = start; seq <= end; seq++) {
        if (this.ackMap.has(seq)) {
          this.clearTimer(seq);
          this.ackMap.delete(seq);
        }
      }
    }
    // 3. 触发发送缓冲区排空回调
    this.drainSendBuffer();
  }

  // 定时器重传
  private startTimer(seq: number): void {
    this.timers.set(seq, setTimeout(() => {
      const pkt = this.ackMap.get(seq);
      if (pkt && pkt.retries < MAX_RETRIES) {
        pkt.retries++;
        this.dc.send(this.encode(pkt));
        this.startTimer(seq); // 重置定时器
      } else {
        this.emit('error', new Error(`Chunk ${pkt?.chunkIndex} max retries exceeded`));
      }
    }, RETRANSMIT_TIMEOUT));
  }
}

// persistence.ts (IndexedDB封装)
const DB_NAME = 'webrtc-file-transfer';
const STORE_PROGRESS = 'progress';
const STORE_CHUNKS = 'chunks'; // 可选：缓存已接收分片，断点恢复加速

export async function saveProgress(manifest: FileManifest, received: BitSet): Promise<void> {
  const db = await openDB();
  await db.put(STORE_PROGRESS, {
    fileId: manifest.fileId,
    manifest,
    receivedBitmap: received.toBase64(), // 压缩存储
    updatedAt: Date.now()
  });
}

export async function loadProgress(fileId: string): Promise<ProgressSnapshot | null> {
  const db = await openDB();
  return db.get(STORE_PROGRESS, fileId);
}

// 恢复逻辑
async function resumeTransfer(fileId: string, dc: RTCDataChannel) {
  const snap = await loadProgress(fileId);
  if (!snap) return startFreshTransfer(fileId);

  // 1. 校验文件是否变更
  const currentHash = await computeFileHash(snap.manifest.filePath);
  if (currentHash !== snap.manifest.sha256) {
    await clearProgress(fileId); // 文件已修改，重新开始
    return startFreshTransfer(fileId);
  }

  // 2. 计算缺失分片
  const missing = snap.manifest.chunks
    .filter((_, i) => !snap.receivedBitmap.get(i))
    .map(c => c.index);

  // 3. 通知对端从缺失分片开始（通过控制通道发送 RESUME 指令）
  dc.send(JSON.stringify({ type: 'RESUME', fileId, missingChunks: missing }));
}

// shared/protocol.ts - 单一事实来源
import { z } from 'zod';

// 1. 信令层契约
export const SignalMessageSchema = z.discriminatedUnion('type', [
  z.object({ type: z.literal('OFFER'), payload: z.object({ sdp: z.string(), fileId: z.string().uuid() }) }),
  z.object({ type: z.literal('ANSWER'), payload: z.object({ sdp: z.string() }) }),
  z.object({ type: z.literal('ICE_CANDIDATE'), payload: z.object({ candidate: z.string(), sdpMid: z.string(), sdpMLineIndex: z.number() }) }),
  z.object({ type: z.literal('RESUME'), payload: z.object({ fileId: z.string().uuid(), missingChunks: z.array(z.number().int().nonnegative()) }) }),
  z.object({ type: z.literal('ERROR'), payload: z.object({ code: z.string(), message: z.string(), recoverable: z.boolean() }) }),
]);
export type SignalMessage = z.infer<typeof SignalMessageSchema>;

// 2. DataChannel 控制帧契约（二进制首部 + JSON Body）
export const ControlFrameHeaderSchema = z.object({
  version: z.literal(1),
  frameType: z.enum(['ACK', 'NACK', 'WINDOW_UPDATE', 'MANIFEST', 'HEARTBEAT']),
  flags: z.number().int().min(0).max(255), // 位域：0x01=END_OF_STREAM, 0x02=PRIORITY
  streamId: z.number().int().nonnegative(), // SCTP Stream ID
  sequence: z.number().int().nonnegative(), // 全局单调递增
  timestamp: z.number().int().nonnegative(), // 发送端 Unix ms
  payloadLength: z.number().int().nonnegative(),
});
export type ControlFrameHeader = z.infer<typeof ControlFrameHeaderSchema>;

// 3. 运行时守卫（编译期通过，运行期再校验）
export function parseControlFrame(view: DataView): ControlFrameHeader | null {
  const raw = {
    version: view.getUint8(0),
    frameType: view.getUint8(1),
    flags: view.getUint8(2),
    streamId: view.getUint16(3, false),
    sequence: view.getUint32(5, false),
    timestamp: view.getBigUint64(9, false),
    payloadLength: view.getUint32(17, false),
  };
  const result = ControlFrameHeaderSchema.safeParse(raw);
  if (!result.success) {
    logger.warn('Control frame parse failed', { errors: result.error.flatten() });
    return null;
  }
  return result.data;
}

test.describe.configure({ retries: 3 }); // 弱网易抖动，允许重试
test('1GB 文件在 2% 丢包 / 100ms RTT 下完成传输且校验通过', async ({ page, browserName }) => {
  await page.route('**/turn/**', route => route.continue()); // 强制走 TURN 模拟中转
  await page.evaluate(() => window.simulateNetwork({ loss: 0.02, rtt: 100 }));
  
  const { fileId, done } = await startTransfer(page, '1GB.bin');
  await expect.poll(() => getProgress(page, fileId), { timeout: 300_000, interval: 5_000 })
    .toEqual(1.0);
  
  const hash = await page.evaluate(fid => window.verifyFileHash(fid), fileId);
  expect(hash).toBe(EXPECTED_SHA256);
});

# .gitlab-ci.yml 片段
stages:
  - lint_type
  - unit_test
  - integration_test
  - e2e_chrome
  - e2e_firefox
  - e2e_safari_webkit # 通过 playwright-webkit 跑 Safari 逻辑
  - chaos_test       # 夜ly 触发，需专用弱网节点
  - bundle_analysis  # webpack-bundle-analyzer 防体积膨胀
  - release

e2e_chrome:
  stage: e2e_chrome
  image: mcr.microsoft.com/playwright:v1.42-jammy
  services:
    - name: registry.infra/turn-server:latest # 自建 TURN
      alias: turn
  variables:
    TURN_URI: "turn:turn:3478?transport=udp"
    TURN_USER: "ci"
    TURN_PASS: "$CI_TURN_SECRET"
  script:
    - pnpm run build:prod
    - pnpm run test:e2e --project=chromium --reporter=junit
  artifacts:
    reports:
      junit: test-results/*.xml
    when: always
    expire_in: 30d

# Cargo.toml
[lib]
name = "reliable_transport_core"
crate-type = ["cdylib", "rlib", "staticlib"]

[dependencies]
webrtc-sctp = { git = "https://github.com/webrtc-rs/webrtc", features = ["datachannel"] } # 复用 SCTP 逻辑
sha2 = "0.10"
bincode = "1.3"
serde = { version = "1.0", features = ["derive"] }
thiserror = "1.0"
# WASM 目标
wasm-bindgen = { version = "0.2", features = ["serde-serialize"] }
# NAPI 目标
napi = { version = "2", features = ["napi5"] }
napi-derive = "2"
# FFI 目标 (Flutter/Dart)
flutter_rust_bridge = "1.72"

use napi_derive::napi;
use flutter_rust_bridge::frb;
use wasm_bindgen::prelude::*;

#[napi] // Node/Electron
#[frb(sync)] // Flutter (同步调用需谨慎，建议异步)
#[wasm_bindgen] // Web Worker
pub struct TransportConfig {
    pub chunk_size: u32,       // 65536
    pub window_size: u16,      // 128
    pub max_retries: u8,       // 5
    pub rto_initial_ms: u32,   // 800
}

#[napi]
#[frb]
#[wasm_bindgen]
impl TransportConfig {
    #[napi(constructor)]
    #[frb(ignore)] // Flutter 用默认值构造
    #[wasm_bindgen(constructor)]
    pub fn new() -> Self { Self::default() }
}

// 核心状态机：发送端
#[napi]
#[frb]
#[wasm_bindgen]
pub struct SenderEngine {
    inner: Arc<Mutex<SenderInner>>, // 线程安全
}

#[napi]
#[frb]
#[wasm_bindgen]
impl SenderEngine {
    #[napi(constructor)]
    #[frb(ignore)]
    #[wasm_bindgen(constructor)]
    pub fn new(config: TransportConfig, file_id: String, manifest: Vec<u8>) -> Self { ... }

    // 供上层调用：获取下一个待发送数据包 (零拷贝)
    #[napi]
    #[frb(async)] // Flutter/Dart 异步
    #[wasm_bindgen(js_name = "nextPacket")]
    pub async fn next_packet(&self) -> Option<Vec<u8>> { ... }

    // 供上层调用：处理 ACK/NACK
    #[napi]
    #[frb(sync)]
    #[wasm_bindgen(js_name = "handleAck")]
    pub fn handle_ack(&self, ack_data: &[u8]) { ... }

    // 供上层调用：定时器驱动重传
    #[napi]
    #[frb(sync)]
    #[wasm_bindgen(js_name = "tick")]
    pub fn tick(&self, now_ms: u64) { ... }
}

graph TD
    A[发送端选文件] --> B{客户端预检}
    B -->|扩展名/魔数/大小| C[通过]
    B -->|可疑| D[拦截/人工复核]
    C --> E[分片 & 计算哈希]
    E --> F[上传 Manifest 至合规网关]
    F --> G{合规网关异步扫描}
    G -->|病毒/敏感词/违规| H[阻断传输/冻结账号/审计日志]
    G -->|通过| I[下发传输许可 Token JWT]
    I --> J[WebRTC 传输开始]
    J --> K[接收端流式校验]
    K -->|分片哈希不符| L[请求重传/标记损坏]
    K -->|全文哈希不符| M[拒绝入库/触发溯源]
    M --> N[审计日志入库]

// receiver-stream-hash.ts
import { createHash } from 'crypto'; // Node / Electron
// 浏览器端用 crypto.subtle.digest 配合 ReadableStream

export class StreamingVerifier {
  private hasher = createHash('sha256');
  private expectedTotal: number;
  private received = 0;
  private readonly expectedFinalHash: string;

  constructor(manifest: FileManifest) {
    this.expectedTotal = manifest.fileSize;
    this.expectedFinalHash = manifest.sha256;
  }

  write(chunk: Buffer | Uint8Array): void {
    this.hasher.update(chunk);
    this.received += chunk.length;
    if (this.received > this.expectedTotal) throw new Error('Size overflow');
  }

  finalize(): boolean {
    if (this.received !== this.expectedTotal) throw new Error('Size mismatch');
    const actual = this.hasher.digest('hex');
    return actual === this.expectedFinalHash;
  }
}

{
  "event": "FILE_TRANSFER_COMPLETED",
  "timestamp": "2024-05-20T10:30:00.123Z",
  "traceId": "01HXYZ...", // W3C TraceContext
  "actor": {
    "userId": "u_123",
    "deviceId": "dev_abc",
    "ip": "203.0.113.45", // 脱敏存储最后一段
    "geo": "CN-SH"
  },
  "resource": {
    "fileId": "f_789",
    "fileNameHash": "sha256:abc...", // 不存明文文件名
    "size": 1073741824,
    "mimeType": "application/pdf"
  },
  "network": {
    "protocol": "WebRTC/DataChannel",
    "icePath": "TURN-UDP", // 或 STUN, RELAY-TCP
    "durationMs": 89200,
    "throughputMbps": 92.4,
    "retransmitRate": 0.012,
    "cryptoSuite": "TLS_AES_256_GCM_SHA384"
  },
  "compliance": {
    "scanResult": "PASS",
    "scanEngine": "ClamAV/0.105 + DLP_RuleSet_v202405",
    "classification": "INTERNAL_CONFIDENTIAL",
    "retentionDays": 2555
  },
  "integrity": {
    "manifestHash": "sha256:...",
    "finalHashVerified": true
  }
}

// tracing.ts - W3C TraceContext 传播
import { propagation, context, trace } from '@opentelemetry/api';

// 1. 发起端：创建 Span 并注入信令
function startTransferTrace(fileId: string) {
  const tracer = trace.getTracer('file-transfer');
  return tracer.startActiveSpan('webrtc.transfer', { attributes: { 'file.id': fileId } }, (span) => {
    // 将 traceparent 注入信令消息
    const carrier: Record<string, string> = {};
    propagation.inject(context.active(), carrier);
    signalSocket.send(JSON.stringify({ type: 'OFFER', traceparent: carrier.traceparent, ... }));
    return span;
  });
}

// 2. 接收端：从信令提取上下文
function handleOffer(msg: SignalMessage) {
  const ctx = propagation.extract(context.active(), { traceparent: msg.traceparent });
  return trace.getTracer('file-transfer').startActiveSpan('webrtc.receive', undefined, ctx, (span) => {
    span.setAttribute('file.id', msg.payload.fileId);
    // 后续 DataChannel 事件挂载为 Child Span
    return span;
  });
}

// 3. 关键节点埋点
dc.onopen = () => span?.addEvent('datachannel_open');
dc.onmessage = (e) => {
  if (isControlFrame(e.data)) {
    const child = tracer.startSpan('sctp.control_frame', undefined, trace.setSpan(context.active(), span));
    child.setAttribute('frame.type', frameType);
    child.end();
  }
};

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: turn-server-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: coturn
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: External
    external:
      metric:
        name: turn_allocate_requests_per_second
      target:
        type: AverageValue
        averageValue: "100" # 每 Pod 每秒处理 100 个分配请求
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 60
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 600 # 缩容冷却 10 分钟，防抖
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 60

// transport-selector.ts
export async function selectTransport(caps: ClientCapabilities): Promise<Transport> {
  // 1. 优先 WebTransport (低延迟、服务端简单)
  if (caps.webtransport && caps.h3) {
    const wt = await tryConnectWebTransport();
    if (wt) return new WebTransportAdapter(wt);
  }
  // 2. 回退 WebRTC (穿透能力强、生态成熟)
  if (caps.webrtc) {
    return new WebRTCAdapter(await createPeerConnection());
  }
  // 3. 兜底 HTTP/2 分片
  return new HttpFallbackAdapter();
}

// wasm/src/media_pipeline.rs
use web_codecs::{VideoDecoder, VideoEncoder, VideoFrame};
use wasm_bindgen::prelude::*;

#[wasm_bindgen]
pub struct MediaPipeline {
    decoder: VideoDecoder,
    encoder: VideoEncoder,
}

#[wasm_bindgen]
impl MediaPipeline {
    #[wasm_bindgen(constructor)]
    pub fn new(config: JsValue) -> Result<MediaPipeline, JsValue> { ... }

    // 输入: 传入分片流 -> 解码 -> 处理 (水印/扫描) -> 编码 -> 输出分片流
    // 关键：VideoFrame 在 WASM 与 JS 间**零拷贝传递** (WASM GC 提案落地后)
    #[wasm_bindgen]
    pub async fn process_chunk(&mut self, chunk: &[u8]) -> Result<Vec<u8>, JsValue> { ... }
}

// sha256.comp.glsl (WebGPU Compute Shader)
@group(0) @binding(0) var<storage, read> input: array<u32>;
@group(0) @binding(1) var<storage, read_write> state: array<u32>; // 8 x u32 = 256bit
@group(0) @binding(2) var<storage, write> output: array<u32>;

@compute @workgroup_size(64)
fn main(@builtin(global_invocation_id) gid: vec3<u32>) {
    // 并行 SHA-256 实现 (参考 RFC 6234)
    // 单次 Dispatch 处理 64 字节块，GPU 并行度极高
    // 适用于：大文件完整性校验、分片哈希批量计算、密钥派生
}